diff --git a/samples/BareWire.Samples.AppHost/Program.cs b/samples/BareWire.Samples.AppHost/Program.cs index 277352e..f29905c 100644 --- a/samples/BareWire.Samples.AppHost/Program.cs +++ b/samples/BareWire.Samples.AppHost/Program.cs @@ -9,16 +9,13 @@ .WithLifetime(ContainerLifetime.Session) .WithManagementPlugin(); -// Enable PostgreSQL prepared transactions (2-phase commit) via max_prepared_transactions. -// System.Transactions.TransactionScope — used by the BareWire transactional outbox middleware — -// escalates to 2PC when a consumer enlists a SECOND database connection inside the consume -// transaction (e.g. the OrderedConsumers sample persists a ProcessedRecord via its own DbContext -// alongside the outbox/inbox writes). PostgreSQL ships with max_prepared_transactions=0 (2PC -// disabled), which makes such a consume abort with SqlState 55000 ("prepared transactions are -// disabled") and dead-letter every message. A small nonzero pool enables the atomic commit. +// No max_prepared_transactions configuration is needed. Every transactional-outbox sample shares the +// outbox middleware's pinned connection for its consumer business write (via IOutboxConnectionAccessor), +// so each consume commits single-phase — no second connection enlists, and System.Transactions.TransactionScope +// never escalates to a two-phase (prepared) commit. PostgreSQL therefore runs with its default +// max_prepared_transactions=0 (2PC disabled). var postgresServer = builder.AddPostgres("postgres") - .WithLifetime(ContainerLifetime.Session) - .WithArgs("-c", "max_prepared_transactions=100"); + .WithLifetime(ContainerLifetime.Session); var postgres = postgresServer.AddDatabase("barewiredb"); diff --git a/samples/BareWire.Samples.InboxDeduplication/Program.cs b/samples/BareWire.Samples.InboxDeduplication/Program.cs index c18ab1f..29504c7 100644 --- a/samples/BareWire.Samples.InboxDeduplication/Program.cs +++ b/samples/BareWire.Samples.InboxDeduplication/Program.cs @@ -72,7 +72,28 @@ // 3. EF Core — application DbContext for notification log entities // ───────────────────────────────────────────────────────────────────────────── -builder.Services.AddDbContext(o => o.UseNpgsql(dbConnectionString)); +// Single-commit connection sharing: while the transactional outbox middleware processes a message it +// has pinned ONE connection (exposed via IOutboxConnectionAccessor). Both consumers below persist a +// NotificationLog through THIS context — share the pinned connection so each write commits single-phase +// with the inbox marker — one physical connection, one enlistment, no escalation to a two-phase +// (prepared) commit, and no max_prepared_transactions requirement. Outside a consume operation (startup +// schema init, the HTTP endpoints below) the accessor returns null, so we fall back to a standalone +// connection. The (sp, options) overload makes the EF options Scoped, so each per-message consumer scope +// binds to the live pinned connection at resolution time. +builder.Services.AddDbContext((sp, o) => +{ + System.Data.Common.DbConnection? sharedOutboxConnection = + sp.GetRequiredService().Current; + + if (sharedOutboxConnection is not null) + { + o.UseNpgsql(sharedOutboxConnection); + } + else + { + o.UseNpgsql(dbConnectionString); + } +}); // ───────────────────────────────────────────────────────────────────────────── // 4. BareWire messaging — serializer, transport, topology, endpoints diff --git a/samples/BareWire.Samples.OrderedConsumers/Program.cs b/samples/BareWire.Samples.OrderedConsumers/Program.cs index 08748f4..81c8a07 100644 --- a/samples/BareWire.Samples.OrderedConsumers/Program.cs +++ b/samples/BareWire.Samples.OrderedConsumers/Program.cs @@ -72,13 +72,34 @@ // 3. EF Core — application DbContext for processed-record log // ───────────────────────────────────────────────────────────────────────────── -// NOTE (PostgreSQL 2PC): the consumers below persist a ProcessedRecord through THIS DbContext while -// the transactional outbox middleware holds an ambient TransactionScope on its own OutboxDbContext. -// That second connection makes the scope escalate to a two-phase (prepared) commit, which requires -// max_prepared_transactions > 0 on PostgreSQL (disabled by default → consume aborts with SqlState -// 55000 and every message is dead-lettered). The Aspire AppHost sets it; a manual Postgres container -// needs `-c max_prepared_transactions=100`. See BareWire.Outbox.EntityFramework/README.md. -builder.Services.AddDbContext(o => o.UseNpgsql(dbConnectionString)); +// SINGLE-COMMIT connection sharing: the consumers below persist a ProcessedRecord through THIS +// DbContext while the transactional outbox middleware holds an ambient TransactionScope on its own +// OutboxDbContext. If this context opened its OWN connection, the scope would enlist two connections +// and escalate to a two-phase (prepared) commit — slower, and disabled by default on PostgreSQL +// (max_prepared_transactions=0 → SqlState 55000, every message dead-lettered). +// +// Instead we share the SINGLE connection the outbox middleware has already pinned for the in-flight +// message (exposed via IOutboxConnectionAccessor). One physical connection → one enlistment → a +// single-phase commit of the business write + outbox messages + inbox marker. No 2PC, no +// max_prepared_transactions requirement. The accessor returns null outside a consume operation +// (startup schema init, the HTTP endpoints below), so we fall back to a standalone connection there. +// +// The (sp, options) overload makes the EF options Scoped — rebuilt per DI scope — so each per-message +// consumer scope reads the accessor at resolution time and binds to the live pinned connection. +builder.Services.AddDbContext((sp, o) => +{ + System.Data.Common.DbConnection? sharedOutboxConnection = + sp.GetRequiredService().Current; + + if (sharedOutboxConnection is not null) + { + o.UseNpgsql(sharedOutboxConnection); + } + else + { + o.UseNpgsql(dbConnectionString); + } +}); // No singleton PoisonKeyHolder needed — the poison-head indicator is stamped as a // transport header ("poison-head-demo: true") on seq=0 of the poison key only. This diff --git a/samples/BareWire.Samples.TransactionalOutbox/Consumers/TransferConsumer.cs b/samples/BareWire.Samples.TransactionalOutbox/Consumers/TransferConsumer.cs index a2368d5..2eab988 100644 --- a/samples/BareWire.Samples.TransactionalOutbox/Consumers/TransferConsumer.cs +++ b/samples/BareWire.Samples.TransactionalOutbox/Consumers/TransferConsumer.cs @@ -32,8 +32,9 @@ public async Task ConsumeAsync(ConsumeContext context) await Task.Delay(millisecondsDelay: 50, context.CancellationToken).ConfigureAwait(false); // Update the transfer status in the same transaction managed by TransactionalOutboxMiddleware. - // The OutboxDbContext and TransferDbContext share the same connection string, - // so both participate in the same ambient TransactionScope. + // This TransferDbContext shares the SAME physical connection the middleware pinned for this + // message (via IOutboxConnectionAccessor — see Program.cs), so the UPDATE commits single-phase + // with the outbox + inbox writes: one connection, one enlistment, no two-phase (prepared) commit. Transfer? transfer = await dbContext.Transfers .FirstOrDefaultAsync(t => t.TransferId == message.TransferId, context.CancellationToken) .ConfigureAwait(false); diff --git a/samples/BareWire.Samples.TransactionalOutbox/Program.cs b/samples/BareWire.Samples.TransactionalOutbox/Program.cs index c675794..d969a2d 100644 --- a/samples/BareWire.Samples.TransactionalOutbox/Program.cs +++ b/samples/BareWire.Samples.TransactionalOutbox/Program.cs @@ -66,7 +66,27 @@ // 3. EF Core — application DbContext for Transfer entities // ───────────────────────────────────────────────────────────────────────────── -builder.Services.AddDbContext(o => o.UseNpgsql(dbConnectionString)); +// Single-commit connection sharing: while the transactional outbox middleware processes a message it +// has pinned ONE connection (exposed via IOutboxConnectionAccessor). Share it so the consumer's UPDATE +// commits single-phase with the outbox + inbox writes — one physical connection, one enlistment, no +// escalation to a two-phase (prepared) commit, and no max_prepared_transactions requirement. Outside a +// consume operation (startup schema init, the HTTP endpoints below) the accessor returns null, so we +// fall back to a standalone connection. The (sp, options) overload makes the EF options Scoped, so each +// per-message consumer scope binds to the live pinned connection at resolution time. +builder.Services.AddDbContext((sp, o) => +{ + System.Data.Common.DbConnection? sharedOutboxConnection = + sp.GetRequiredService().Current; + + if (sharedOutboxConnection is not null) + { + o.UseNpgsql(sharedOutboxConnection); + } + else + { + o.UseNpgsql(dbConnectionString); + } +}); // ───────────────────────────────────────────────────────────────────────────── // 4. BareWire messaging — serializer, transport, topology, endpoints diff --git a/samples/README.md b/samples/README.md index 47bd3cd..6b72b10 100644 --- a/samples/README.md +++ b/samples/README.md @@ -18,16 +18,17 @@ If you prefer to run a sample individually, start RabbitMQ and PostgreSQL first: ```bash docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:management -docker run -d --name postgres -p 5432:5432 -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=barewiredb postgres -c max_prepared_transactions=100 +docker run -d --name postgres -p 5432:5432 -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=barewiredb postgres ``` -> **Why `-c max_prepared_transactions=100`?** The transactional-outbox samples wrap each consume in a -> `System.Transactions.TransactionScope`. When a consumer also persists business state through **its own** -> `DbContext` (a second database connection) — as `OrderedConsumers` does with its `ProcessedRecord` log — -> the scope escalates to a two-phase (prepared) commit. PostgreSQL ships with `max_prepared_transactions=0` -> (2PC disabled), so without this flag those consumes abort with `55000: prepared transactions are disabled` -> and every message is dead-lettered. The Aspire AppHost sets this automatically; a manual container needs -> it explicitly. See the [transactional outbox limitation](../src/BareWire.Outbox.EntityFramework/README.md). +> **No `max_prepared_transactions` needed.** A consumer that persists its own business state while the +> transactional outbox middleware holds an ambient `System.Transactions.TransactionScope` would normally +> open a **second** database connection, escalating the commit to a two-phase (prepared) commit — which +> PostgreSQL disables by default (`max_prepared_transactions=0` → `55000: prepared transactions are +> disabled`). Every outbox sample here avoids that by **sharing the outbox's pinned connection** for the +> consumer write (via `IOutboxConnectionAccessor`), so each consume commits **single-phase**: faster, and +> with no `max_prepared_transactions` requirement. See the +> [single-commit vs 2PC guidance](../src/BareWire.Outbox.EntityFramework/README.md). Then run the sample: @@ -132,6 +133,11 @@ GET /events/processing-log — verify per-correlation ordering End-to-end per-key consumer ordering with competing consumer instances (Aspire `WithReplicas(2)`), transactional outbox (`OrderingMode.PerKey`), and poison-head parking via DLX. +The consumers persist a `ProcessedRecord` through the **same connection the outbox middleware pinned** +for the in-flight message (via `IOutboxConnectionAccessor`), so the business write commits **single-phase** +with the outbox and inbox writes — one commit, no two-phase (prepared) commit, and no +`max_prepared_transactions` requirement on PostgreSQL. + Demonstrates two ordering tiers (ADR-026): 1. **Cross-instance (SAC)** — `ordered-processing` queue with `x-single-active-consumer`. RabbitMQ diff --git a/src/BareWire.Outbox.EntityFramework/IOutboxConnectionAccessor.cs b/src/BareWire.Outbox.EntityFramework/IOutboxConnectionAccessor.cs new file mode 100644 index 0000000..a97f48b --- /dev/null +++ b/src/BareWire.Outbox.EntityFramework/IOutboxConnectionAccessor.cs @@ -0,0 +1,47 @@ +using System.Data.Common; + +namespace BareWire.Outbox.EntityFramework; + +/// +/// Exposes the database connection that the transactional outbox middleware has pinned for the +/// message currently being consumed on the active asynchronous flow. +/// +/// +/// +/// The transactional outbox middleware opens a single physical connection for the lifetime of a +/// consume operation and enlists it once in the ambient transaction. A consumer can persist its +/// own business state through that same connection — instead of opening a second one — so +/// that the business write, the outbox messages, and the inbox processed marker all commit as a +/// single-phase commit. Sharing one connection avoids escalation to a two-phase (prepared) commit, +/// which is both faster and free of the PostgreSQL max_prepared_transactions requirement. +/// +/// +/// Typical usage is to configure a consumer's DbContext to use when it +/// is non- and fall back to its own connection otherwise (startup schema +/// initialization, HTTP request handlers, or any path that runs outside a consume operation): +/// +/// +/// services.AddDbContext<MyDbContext>((sp, options) => +/// { +/// DbConnection? shared = sp.GetRequiredService<IOutboxConnectionAccessor>().Current; +/// if (shared is not null) +/// options.UseNpgsql(shared); // share the outbox connection → single commit +/// else +/// options.UseNpgsql(connectionString); // standalone connection +/// }); +/// +/// +/// The accessor is registered as a singleton by . +/// It is backed by an asynchronous-flow-local value, so reflects the +/// connection pinned by the outbox middleware on the caller's logical execution context. +/// +/// +public interface IOutboxConnectionAccessor +{ + /// + /// Gets the open the transactional outbox middleware has pinned for + /// the in-flight consume operation on the current asynchronous flow, or + /// when no outbox consume operation is in progress. + /// + DbConnection? Current { get; } +} diff --git a/src/BareWire.Outbox.EntityFramework/Internal/OutboxConnectionAccessor.cs b/src/BareWire.Outbox.EntityFramework/Internal/OutboxConnectionAccessor.cs new file mode 100644 index 0000000..c179c87 --- /dev/null +++ b/src/BareWire.Outbox.EntityFramework/Internal/OutboxConnectionAccessor.cs @@ -0,0 +1,13 @@ +using System.Data.Common; + +namespace BareWire.Outbox.EntityFramework.Internal; + +/// +/// Default implementation. Reads the connection pinned by +/// on the current asynchronous flow. Stateless and +/// thread-safe — safe to register as a singleton. +/// +internal sealed class OutboxConnectionAccessor : IOutboxConnectionAccessor +{ + public DbConnection? Current => TransactionalOutboxMiddleware.CurrentConnection; +} diff --git a/src/BareWire.Outbox.EntityFramework/README.md b/src/BareWire.Outbox.EntityFramework/README.md index e6447dd..f929de1 100644 --- a/src/BareWire.Outbox.EntityFramework/README.md +++ b/src/BareWire.Outbox.EntityFramework/README.md @@ -32,26 +32,51 @@ The inbox `ProcessedAt` marker is committed **atomically** within the same trans consumer's business state and buffered outbox messages, so a crash after the business commit cannot leave a message reprocessable (see ADR-033). -### PostgreSQL: prepared transactions (2PC) requirement +### PostgreSQL: consumer business writes — single-commit vs 2PC The atomic commit above uses a `System.Transactions.TransactionScope`. The middleware pins **one** -physical connection for its own inbox/outbox writes, so the common case stays single-connection (no -two-phase commit). However, if your **consumer also persists business state through a separate -`DbContext`** (a second database connection enlisted in the same scope), `TransactionScope` escalates -to a **two-phase (prepared) commit**. PostgreSQL ships with `max_prepared_transactions = 0` (prepared -transactions disabled), so such a consume aborts with `55000: prepared transactions are disabled` and -the message is retried until dead-lettered. - -To support a consumer that writes to its own `DbContext` under the transactional outbox on PostgreSQL, -either: - -- set a nonzero `max_prepared_transactions` on the server (e.g. start Postgres with - `-c max_prepared_transactions=100`), enabling 2PC; or -- have the consumer persist through a connection **shared** with `OutboxDbContext` (single connection → - no escalation). - -The BareWire samples take the first option — the Aspire AppHost configures -`max_prepared_transactions` automatically (see `samples/README.md`). +physical connection for its own inbox/outbox writes, so the common case stays single-connection. But a +frequent pattern is for the **consumer to also persist business state through its own `DbContext`** +inside the same transaction. How that second write enlists decides whether the commit is one phase or +two: + +- **Two physical connections → two-phase (prepared) commit.** If the consumer's `DbContext` opens its + own connection, `TransactionScope` enlists two resources and escalates to a 2PC. PostgreSQL ships with + `max_prepared_transactions = 0` (prepared transactions disabled), so the consume aborts with + `55000: prepared transactions are disabled` and the message is retried until dead-lettered. Enabling it + (start Postgres with `-c max_prepared_transactions=100`) makes it work, but a prepared commit is also + **slower** — an extra `PREPARE` / `COMMIT PREPARED` round-trip and fsync per message. + +- **One shared connection → single-phase commit (recommended).** Have the consumer's `DbContext` use the + **same** connection the middleware already pinned for the in-flight message, exposed via + `IOutboxConnectionAccessor`. One physical connection enlists exactly once, so the business write, the + buffered outbox messages, and the inbox marker all commit in a single local transaction — **faster**, + and with **no** `max_prepared_transactions` requirement at all. + +Wire the consumer's `DbContext` to prefer the shared connection, falling back to a standalone connection +outside a consume operation (startup schema creation, HTTP request handlers, background jobs): + +```csharp +services.AddDbContext((sp, options) => +{ + // System.Data.Common.DbConnection — non-null only while the outbox middleware is processing + // a message on the current async flow; null on startup / HTTP / background paths. + DbConnection? shared = sp.GetRequiredService().Current; + if (shared is not null) + options.UseNpgsql(shared); // share the outbox connection → single-phase commit + else + options.UseNpgsql(connectionString); // standalone connection +}); +``` + +> Use the `(IServiceProvider, DbContextOptionsBuilder)` overload so EF builds the options **per scope** — +> each per-message consumer scope then binds to the live pinned connection. The consumer keeps calling +> `SaveChangesAsync()` as usual; because it runs inside the middleware's `TransactionScope`, its write +> commits atomically with the outbox and inbox writes — now as one single-phase commit. + +Every BareWire outbox sample whose consumer persists business state — `OrderedConsumers`, +`TransactionalOutbox`, `InboxDeduplication` — uses this single-commit pattern, so the samples need no 2PC +and the Aspire AppHost runs PostgreSQL with the default `max_prepared_transactions = 0`. ## Horizontal Scaling and Row Claims diff --git a/src/BareWire.Outbox.EntityFramework/ServiceCollectionExtensions.cs b/src/BareWire.Outbox.EntityFramework/ServiceCollectionExtensions.cs index 8d305be..f4d5b89 100644 --- a/src/BareWire.Outbox.EntityFramework/ServiceCollectionExtensions.cs +++ b/src/BareWire.Outbox.EntityFramework/ServiceCollectionExtensions.cs @@ -58,6 +58,11 @@ public static IServiceCollection AddBareWireOutbox( services.AddSingleton(instanceId); + // Exposes the connection the transactional middleware pins per consume operation so a consumer + // DbContext can share it (single-phase commit instead of a two-phase prepared commit). Stateless + // singleton over an async-flow-local — see IOutboxConnectionAccessor for the consumer wiring. + services.TryAddSingleton(); + // Default outbox claim dialect: PostgreSQL (FOR UPDATE SKIP LOCKED). The store invokes a // dialect only when its IOutboxSqlDialect.ProviderName matches the active EF Core provider, // so this default is used on PostgreSQL and is inert elsewhere. To get an atomic claim on diff --git a/src/BareWire.Outbox.EntityFramework/TransactionalOutboxMiddleware.cs b/src/BareWire.Outbox.EntityFramework/TransactionalOutboxMiddleware.cs index 96df15f..f83d3c6 100644 --- a/src/BareWire.Outbox.EntityFramework/TransactionalOutboxMiddleware.cs +++ b/src/BareWire.Outbox.EntityFramework/TransactionalOutboxMiddleware.cs @@ -1,3 +1,4 @@ +using System.Data.Common; using System.Transactions; using BareWire.Abstractions.Pipeline; using Microsoft.EntityFrameworkCore; @@ -14,6 +15,12 @@ internal sealed partial class TransactionalOutboxMiddleware : IMessageMiddleware private static readonly AsyncLocal _current = new(); + // The physical connection pinned for the in-flight consume operation, flowed across the + // consumer's (separate) DI scope via the async execution context. A consumer DbContext can + // share this exact connection so its business write commits single-phase with the outbox + // write and the inbox marker — no second connection, no escalation to a two-phase commit. + private static readonly AsyncLocal _currentConnection = new(); + private readonly OutboxDbContext _dbContext; private readonly IOutboxStore _outboxStore; private readonly InboxFilter _inboxFilter; @@ -21,6 +28,8 @@ internal sealed partial class TransactionalOutboxMiddleware : IMessageMiddleware internal static OutboxBuffer? Current => _current.Value; + internal static DbConnection? CurrentConnection => _currentConnection.Value; + internal TransactionalOutboxMiddleware( OutboxDbContext dbContext, IOutboxStore outboxStore, @@ -59,6 +68,12 @@ public async Task InvokeAsync(MessageContext context, NextMiddleware nextMiddlew // (ExecuteUpdateAsync) share ONE physical connection enlisted once in the ambient // TransactionScope — preventing DTC escalation on Npgsql / non-Windows hosts. await _dbContext.Database.OpenConnectionAsync(ct).ConfigureAwait(false); + + // Publish the pinned connection on the async flow so a consumer DbContext (resolved in its + // own DI scope, but on the same execution context) can share this exact connection. Sharing + // one connection lets the business write commit single-phase with the outbox messages and the + // inbox marker — no second connection, hence no escalation to a two-phase (prepared) commit. + _currentConnection.Value = _dbContext.Database.GetDbConnection(); try { // 1. Inbox deduplication check — deliberately OUTSIDE the TransactionScope so the @@ -130,6 +145,10 @@ await _inboxFilter.MarkProcessedAsync(context.MessageId, consumerType, ct) } finally { + // Stop exposing the pinned connection before it is closed — it must never leak to a + // subsequent message processed on this asynchronous flow. + _currentConnection.Value = null; + // Close the pinned connection on every path. Guard the close so a connection-close // failure never masks the original exception propagating from the try block (handler // fault or commit error) — the real cause must reach the transport for correct settlement. diff --git a/tests/BareWire.UnitTests/Outbox/TransactionalOutboxMiddlewareTests.cs b/tests/BareWire.UnitTests/Outbox/TransactionalOutboxMiddlewareTests.cs index a5bc40a..2fc8278 100644 --- a/tests/BareWire.UnitTests/Outbox/TransactionalOutboxMiddlewareTests.cs +++ b/tests/BareWire.UnitTests/Outbox/TransactionalOutboxMiddlewareTests.cs @@ -3,10 +3,13 @@ #pragma warning disable CA2012 using System.Buffers; +using System.Data; +using System.Data.Common; using AwesomeAssertions; using BareWire.Abstractions.Pipeline; using BareWire.Outbox; using BareWire.Outbox.EntityFramework; +using BareWire.Outbox.EntityFramework.Internal; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; @@ -76,8 +79,101 @@ private static (TransactionalOutboxMiddleware Middleware, IInboxStore InboxStore return (middleware, inboxStore); } + /// + /// Same as but also returns the so a + /// test can assert the connection the middleware pins for sharing is that context's own connection. + /// + private static (TransactionalOutboxMiddleware Middleware, OutboxDbContext DbContext) + CreateMiddlewareWithDbContext() + { + IInboxStore inboxStore = Substitute.For(); + inboxStore + .TryLockAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(ValueTask.FromResult(true)); + inboxStore + .MarkProcessedAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(ValueTask.CompletedTask); + + InboxFilter inboxFilter = new(inboxStore, OutboxOptions.Default, NullLogger.Instance); + + DbContextOptions dbOptions = new DbContextOptionsBuilder() + .UseSqlite("DataSource=:memory:") + .Options; + OutboxDbContext dbContext = new(dbOptions); + + IOutboxStore outboxStore = Substitute.For(); + + TransactionalOutboxMiddleware middleware = new( + dbContext, + outboxStore, + inboxFilter, + NullLogger.Instance); + + return (middleware, dbContext); + } + // ── Tests ───────────────────────────────────────────────────────────────── + /// + /// Regression guard for single-commit connection sharing: while the consumer handler runs, the + /// outbox middleware must expose its pinned, open connection through + /// — and it must be the 's own connection, so a consumer DbContext can + /// share that exact connection and commit single-phase (no escalation to a two-phase commit). + /// + [Fact] + public async Task InvokeAsync_WhileHandlerRuns_ExposesPinnedOpenConnectionViaAccessor() + { + // Arrange + var (middleware, dbContext) = CreateMiddlewareWithDbContext(); + MessageContext context = CreateContext(); + + OutboxConnectionAccessor accessor = new(); + + DbConnection? capturedDuringHandler = null; + ConnectionState capturedState = ConnectionState.Closed; + NextMiddleware next = _ => + { + capturedDuringHandler = accessor.Current; + capturedState = capturedDuringHandler?.State ?? ConnectionState.Closed; + return Task.CompletedTask; + }; + + // Act + await middleware.InvokeAsync(context, next); + + // Assert — during the handler the accessor exposes the middleware's pinned connection, + // it is the OutboxDbContext's own connection, and it is open for the transaction's duration. + capturedDuringHandler.Should().NotBeNull( + "the outbox middleware must expose its pinned connection so a consumer can share it"); + capturedDuringHandler.Should().BeSameAs( + dbContext.Database.GetDbConnection(), + "the exposed connection must be the OutboxDbContext's own connection for single-commit sharing"); + capturedState.Should().Be( + ConnectionState.Open, + "the exposed connection must be open for the duration of the consume transaction"); + } + + /// + /// The pinned connection must not leak past the consume flow: once + /// returns, the accessor reports on this asynchronous flow. + /// + [Fact] + public async Task InvokeAsync_AfterCompletion_ClearsExposedConnection() + { + // Arrange + var (middleware, _) = CreateMiddlewareWithDbContext(); + MessageContext context = CreateContext(); + OutboxConnectionAccessor accessor = new(); + NextMiddleware next = _ => Task.CompletedTask; + + // Act + await middleware.InvokeAsync(context, next); + + // Assert + accessor.Current.Should().BeNull( + "the pinned connection must be cleared after the consume operation completes"); + } + [Fact] public async Task InvokeAsync_WhenDuplicateDetected_SetsInboxFilteredFlag() {