Add transaction configurability to EF saga repository (MassTransit#6130)

* Add transaction configurability to EF saga repository

Introduce `_isTransactionEnabled` to control transaction usage in
`EntityFrameworkSagaRepositoryConfigurator<TSaga>`. Add the
`ConfigureTransaction` method to enable or disable transactions.

Update `CreateOptimisticLockStrategy` and `CreatePessimisticLockStrategy`
to propagate transaction settings. Modify `EntityFrameworkSagaRepository`
methods to accept `isTransactionEnabled` as a parameter.

Enhance `ISagaRepositoryLockStrategy<TSaga>` with `IsTransactionEnabled`
property and update `OptimisticSagaRepositoryLockStrategy` and
`PessimisticSagaRepositoryLockStrategy` to support transaction control.

Refactor `WithinTransaction` in `EntityFrameworkSagaRepositoryContextFactory`
to skip transaction wrapping when disabled. Perform minor refactoring and
import adjustments for consistency.

* ConfigureTransaction method moved to IEntityFrameworkSagaRepositoryConfigurator

* Enforce transactions for pessimistic concurrency

Refactored `CreatePessimisticLockStrategy` and removed the
`isTransactionEnabled` parameter, enforcing transactions for
pessimistic concurrency. Renamed `ConfigureTransaction` to
`SetOptimisticConcurrency` to support optimistic concurrency
with optional transaction control.

Updated `PessimisticSagaRepositoryLockStrategy` to always
enable transactions. Simplified `CreatePessimistic` method
and adjusted related tests to reflect the new design.

Removed tests and configurations for pessimistic concurrency
without transactions, as this is no longer supported. Updated
documentation and test cases to align with the new API.

* Refactor saga transaction tests for clarity and coverage

Refactored test classes to separate optimistic and pessimistic
concurrency tests with transactions enabled/disabled. Updated
test names, descriptions, and categories for better organization.

Consolidated saga repository initialization logic, removing
redundant fields and simplifying setup. Added new test for
handling multiple saga instances without transactions.

Introduced setup and teardown methods to ensure a clean
database state for each test run. Updated lock strategy tests
to reflect transaction-enabled behavior. Removed redundant
code and improved overall maintainability of the test suite.

Author: damir-usmanov

src/Persistence/MassTransit.EntityFrameworkCoreIntegration/Configuration/Configuration/EntityFrameworkSagaRepositoryConfigurator.cs

@@ -25,6 +25,7 @@ public class EntityFrameworkSagaRepositoryConfigurator<TSaga> :
         IsolationLevel _isolationLevel;
         ILockStatementProvider _lockStatementProvider;
         Func<IQueryable<TSaga>, IQueryable<TSaga>> _queryCustomization;
+        bool _isTransactionEnabled = true;
 
         public EntityFrameworkSagaRepositoryConfigurator()
         {
@@ -152,7 +153,7 @@ ISagaRepositoryLockStrategy<TSaga> CreateOptimisticLockStrategy()
         {
             var queryExecutor = new OptimisticLoadQueryExecutor<TSaga>(_queryCustomization);
 
-            return new OptimisticSagaRepositoryLockStrategy<TSaga>(queryExecutor, _queryCustomization, _isolationLevel);
+            return new OptimisticSagaRepositoryLockStrategy<TSaga>(queryExecutor, _queryCustomization, _isolationLevel, _isTransactionEnabled);
         }
 
         ISagaRepositoryLockStrategy<TSaga> CreatePessimisticLockStrategy()
@@ -163,5 +164,11 @@ ISagaRepositoryLockStrategy<TSaga> CreatePessimisticLockStrategy()
 
             return new PessimisticSagaRepositoryLockStrategy<TSaga>(queryExecutor, _isolationLevel);
         }
+
+        public void SetOptimisticConcurrency(bool useTransaction = true)
+        {
+            _concurrencyMode = ConcurrencyMode.Optimistic;
+            _isTransactionEnabled = useTransaction;
+        }
     }
 }

src/Persistence/MassTransit.EntityFrameworkCoreIntegration/Configuration/IEntityFrameworkSagaRepositoryConfigurator.cs

@@ -1,10 +1,10 @@
 namespace MassTransit
 {
+    using EntityFrameworkCoreIntegration;
+    using Microsoft.EntityFrameworkCore;
     using System;
     using System.Data;
     using System.Linq;
-    using EntityFrameworkCoreIntegration;
-    using Microsoft.EntityFrameworkCore;
 
 
     public interface IEntityFrameworkSagaRepositoryConfigurator
@@ -42,6 +42,15 @@ void AddDbContext<TContext, TImplementation>(Action<IServiceProvider, DbContextO
         /// <typeparam name="TContext"></typeparam>
         void ExistingDbContext<TContext>()
             where TContext : DbContext;
+
+        /// <summary>
+        /// Configures the saga to use optimistic concurrency, with optional transaction support.
+        /// </summary>
+        /// <param name="useTransaction">
+        /// If <c>true</c>, operations on the saga will be executed within a transaction; 
+        /// if <c>false</c>, no transaction will be used.
+        /// </param>
+        void SetOptimisticConcurrency(bool useTransaction = true);
     }
 
 

src/Persistence/MassTransit.EntityFrameworkCoreIntegration/EntityFrameworkCoreIntegration/EntityFrameworkSagaRepository.cs

@@ -12,18 +12,18 @@ public static class EntityFrameworkSagaRepository<TSaga>
         where TSaga : class, ISaga
     {
         public static ISagaRepository<TSaga> CreateOptimistic(ISagaDbContextFactory<TSaga> dbContextFactory,
-            Func<IQueryable<TSaga>, IQueryable<TSaga>> queryCustomization = null)
+            Func<IQueryable<TSaga>, IQueryable<TSaga>> queryCustomization = null, bool isTransactionEnabled = true)
         {
             var queryExecutor = new OptimisticLoadQueryExecutor<TSaga>(queryCustomization);
-            var lockStrategy = new OptimisticSagaRepositoryLockStrategy<TSaga>(queryExecutor, queryCustomization, IsolationLevel.ReadCommitted);
+            var lockStrategy = new OptimisticSagaRepositoryLockStrategy<TSaga>(queryExecutor, queryCustomization, IsolationLevel.ReadCommitted, isTransactionEnabled);
 
             return CreateRepository(dbContextFactory, lockStrategy);
         }
 
         public static ISagaRepository<TSaga> CreateOptimistic(Func<DbContext> dbContextFactory,
-            Func<IQueryable<TSaga>, IQueryable<TSaga>> queryCustomization = null)
+            Func<IQueryable<TSaga>, IQueryable<TSaga>> queryCustomization = null, bool isTransactionEnabled = true)
         {
-            return CreateOptimistic(new DelegateSagaDbContextFactory<TSaga>(dbContextFactory), queryCustomization);
+            return CreateOptimistic(new DelegateSagaDbContextFactory<TSaga>(dbContextFactory), queryCustomization, isTransactionEnabled);
         }
 
         public static ISagaRepository<TSaga> CreatePessimistic(ISagaDbContextFactory<TSaga> dbContextFactory,

src/Persistence/MassTransit.EntityFrameworkCoreIntegration/EntityFrameworkCoreIntegration/ISagaRepositoryLockStrategy.cs

@@ -1,11 +1,11 @@
 namespace MassTransit.EntityFrameworkCoreIntegration
 {
+    using Microsoft.EntityFrameworkCore;
+    using Saga;
     using System;
     using System.Data;
     using System.Threading;
     using System.Threading.Tasks;
-    using Microsoft.EntityFrameworkCore;
-    using Saga;
 
 
     public interface ISagaRepositoryLockStrategy<TSaga>
@@ -16,5 +16,7 @@ public interface ISagaRepositoryLockStrategy<TSaga>
         Task<TSaga> Load(DbContext context, Guid correlationId, CancellationToken cancellationToken);
 
         Task<SagaLockContext<TSaga>> CreateLockContext(DbContext context, ISagaQuery<TSaga> query, CancellationToken cancellationToken);
+
+        bool IsTransactionEnabled { get; }
     }
 }

src/Persistence/MassTransit.EntityFrameworkCoreIntegration/EntityFrameworkCoreIntegration/Saga/EntityFrameworkSagaRepositoryContextFactory.cs

@@ -1,13 +1,13 @@
 namespace MassTransit.EntityFrameworkCoreIntegration.Saga
 {
+    using MassTransit.Saga;
+    using Microsoft.EntityFrameworkCore;
+    using Microsoft.EntityFrameworkCore.Storage;
     using System;
     using System.Collections.Generic;
     using System.Linq;
     using System.Threading;
     using System.Threading.Tasks;
-    using MassTransit.Saga;
-    using Microsoft.EntityFrameworkCore;
-    using Microsoft.EntityFrameworkCore.Storage;
 
 
     public class EntityFrameworkSagaRepositoryContextFactory<TSaga> :
@@ -161,6 +161,9 @@ Task<T> ExecuteAsync()
 
         Task WithinTransaction(DbContext context, CancellationToken cancellationToken, Func<Task> callback)
         {
+            if (!_lockStrategy.IsTransactionEnabled)
+                return callback();
+
             async Task<bool> Create()
             {
                 await callback().ConfigureAwait(false);

src/Persistence/MassTransit.EntityFrameworkCoreIntegration/EntityFrameworkCoreIntegration/Saga/OptimisticSagaRepositoryLockStrategy.cs

@@ -16,16 +16,19 @@ public class OptimisticSagaRepositoryLockStrategy<TSaga> :
         readonly Func<IQueryable<TSaga>, IQueryable<TSaga>> _queryCustomization;
 
         public OptimisticSagaRepositoryLockStrategy(ILoadQueryExecutor<TSaga> executor, Func<IQueryable<TSaga>, IQueryable<TSaga>> queryCustomization,
-            IsolationLevel isolationLevel)
+            IsolationLevel isolationLevel, bool isTransactionEnabled)
         {
             _executor = executor;
             _queryCustomization = queryCustomization;
 
             IsolationLevel = isolationLevel;
+            IsTransactionEnabled = isTransactionEnabled;
         }
 
         public IsolationLevel IsolationLevel { get; }
 
+        public bool IsTransactionEnabled { get; }
+
         public Task<TSaga> Load(DbContext context, Guid correlationId, CancellationToken cancellationToken)
         {
             return _executor.Load(context, correlationId, cancellationToken);

src/Persistence/MassTransit.EntityFrameworkCoreIntegration/EntityFrameworkCoreIntegration/Saga/PessimisticSagaRepositoryLockStrategy.cs

@@ -24,6 +24,11 @@ public PessimisticSagaRepositoryLockStrategy(ILoadQueryExecutor<TSaga> executor,
 
         public IsolationLevel IsolationLevel { get; }
 
+        /// <summary>
+        /// Pessimistic concurrency always uses transactions as locks require transaction scope.
+        /// </summary>
+        public bool IsTransactionEnabled => true;
+
         public Task<TSaga> Load(DbContext context, Guid correlationId, CancellationToken cancellationToken)
         {
             return _executor.Load(context, correlationId, cancellationToken);

tests/MassTransit.EntityFrameworkCoreIntegration.Tests/TransactionConfiguration/Configurator_TransactionConfiguration_Specs.cs

@@ -0,0 +1,58 @@
+namespace MassTransit.EntityFrameworkCoreIntegration.Tests.TransactionConfiguration
+{
+    using System;
+    using System.Data;
+    using MassTransit.Configuration;
+    using MassTransit.Tests.Saga;
+    using NUnit.Framework;
+
+
+    /// <summary>
+    /// Tests for EntityFrameworkSagaRepositoryConfigurator transaction configuration functionality
+    /// </summary>
+    [TestFixture]
+    public class Configurator_TransactionConfiguration_Specs
+    {
+        [Test]
+        public void Should_accept_configure_transaction_false()
+        {
+            var configurator = new EntityFrameworkSagaRepositoryConfigurator<SimpleSaga>();
+            
+            Assert.DoesNotThrow(() => configurator.SetOptimisticConcurrency(false),
+                "ConfigureTransaction(false) should not throw");
+        }
+
+        [Test]
+        public void Should_accept_configure_transaction_true()
+        {
+            var configurator = new EntityFrameworkSagaRepositoryConfigurator<SimpleSaga>();
+            
+            Assert.DoesNotThrow(() => configurator.SetOptimisticConcurrency(true),
+                "ConfigureTransaction(true) should not throw");
+        }
+
+        [Test]
+        public void Should_allow_multiple_configuration_calls()
+        {
+            var configurator = new EntityFrameworkSagaRepositoryConfigurator<SimpleSaga>();
+            
+            Assert.DoesNotThrow(() => {
+                configurator.SetOptimisticConcurrency(false);
+                configurator.SetOptimisticConcurrency(true);
+                configurator.SetOptimisticConcurrency(false);
+            }, "Multiple ConfigureTransaction calls should not throw");
+        }
+
+        [Test]
+        public void Should_support_fluent_configuration()
+        {
+            var configurator = new EntityFrameworkSagaRepositoryConfigurator<SimpleSaga>();
+            
+            Assert.DoesNotThrow(() => {
+                configurator.ConcurrencyMode = ConcurrencyMode.Optimistic;
+                configurator.SetOptimisticConcurrency(false);
+                configurator.IsolationLevel = IsolationLevel.ReadCommitted;
+            }, "Fluent configuration with transaction setting should work");
+        }
+    }
+}