事务中并行执行任务时的集成测试错误排查

我编写了一个集成测试，通过两个线程同时插入两个文档，期望测试结束后数据库中不存在测试插入的任何记录。我采用事务回滚的方式实现测试完成后回滚所有事务，测试代码使用线程池并行执行任务，基类通过TransactionScope包裹每个测试并在结束时回滚。

测试代码如下：

[TestClass]
public class MyIntegrationTest : IntegrationTestsBase {

    [TestMethod]
    public void SaveTwoDocumentsSimultaneously_WorkSuccessfully()
    {
       //Assign
       var doc1 = new Document() {Number = "Test1"};
       var doc2 = new Document() {Number = "Test2"};

       //action
       CountdownEvent countdown = new CountdownEvent(2);
       ThreadPool.QueueUserWorkItem(WorkerThread, new object[] { Transaction.Current.DependentClone(DependentCloneOption.BlockCommitUntilComplete), doc1, countdown });
       ThreadPool.QueueUserWorkItem(WorkerThread, new object[] { Transaction.Current.DependentClone(DependentCloneOption.BlockCommitUntilComplete), doc2, countdown });
       countdown.Wait();
    
       //assert
       //assertion code for check two document inserted
       ....
    }
}

集成测试基类：

[TestClass]
public abstract class IntegrationTestsBase
{
    private TransactionScope _scope;

    [TestInitialize]
    public void Setup()
    {
       this._scope = new TransactionScope(TransactionScopeOption.Required,
        new System.TimeSpan(0, 10, 0));
    }

    [TestCleanup]
    public void Cleanup()
    {
       this._scope.Dispose();
    } 
}

WorkerThread代码：

private static void WorkerThread(object state)
{
    if (state is object[] array)
    {
        var transaction = array[0];
        var document = array[1] as Document;
        CountdownEvent countdown = array[2] as CountdownEvent;

        try
        {
            //Create a DependentTransaction from the object passed to the WorkerThread
            DependentTransaction dTx = (DependentTransaction)transaction;

            //Sleep for 1 second to force the worker thread to delay
            Thread.Sleep(1000);
            //Pass the DependentTransaction to the scope, so that work done in the scope becomes part of the transaction passed to the worker thread
            using (TransactionScope ts = new TransactionScope(dTx))
            {
                //Perform transactional work here.
                using (var ctx = new PlanningDbContext())
                {
                    ctx.Documents.Add(doc);
                    ctx.SaveChanges();  //<----exception occurs here when second document insert
                }
                //Call complete on the transaction scope
                ts.Complete();
            }

            //Call complete on the dependent transaction
            dTx.Complete();
       }
       catch (Exception ex)
       {
           Debug.WriteLine(ex);
       }
       finally
       {
          countdown?.Signal();
       }
   }
}

运行测试时，第二个文档保存时抛出错误：

System.Data.Entity.Infrastructure.DbUpdateException: An error occurred while updating the entries. See the inner exception for details. ---> System.Data.Entity.Core.UpdateException: An error occurred while updating the entries. See the inner exception for details. ---> System.Transactions.TransactionException: The operation is not valid for the state of the transaction.

1. 事务上下文跨线程传递缺失配置：Transaction.Current是线程静态变量，默认情况下事务上下文无法自动流向线程池线程，导致工作线程中的事务状态失效。
2. 变量名错误：WorkerThread中存在笔误，ctx.Documents.Add(doc)应该是ctx.Documents.Add(document)，这会导致空引用或错误的对象插入，间接触发事务状态异常。
3. DependentTransaction未正确释放：代码中未显式释放DependentTransaction，可能导致事务资源泄漏，引发状态异常。
4. 事务操作顺序问题：虽然代码顺序看似正确，但未确保DependentTransaction在使用完毕后及时完成并释放，加剧了事务状态的不稳定。

1. 启用事务上下文跨线程流 #

在基类创建TransactionScope时，添加TransactionScopeAsyncFlowOption.Enabled参数，确保事务上下文能正确流向线程池线程：

[TestInitialize]
public void Setup()
{
    this._scope = new TransactionScope(
        TransactionScopeOption.Required,
        new System.TimeSpan(0, 10, 0),
        TransactionScopeAsyncFlowOption.Enabled); // 启用跨线程事务流
}

2. 修正WorkerThread代码 #

修复变量名错误，确保DependentTransaction正确释放：

private static void WorkerThread(object state)
{
    if (state is object[] array)
    {
        var dTx = array[0] as DependentTransaction;
        var document = array[1] as Document;
        CountdownEvent countdown = array[2] as CountdownEvent;

        try
        {
            if (dTx == null || document == null) return;

            using (TransactionScope ts = new TransactionScope(dTx))
            {
                using (var ctx = new PlanningDbContext())
                {
                    ctx.Documents.Add(document); // 修正变量名错误
                    ctx.SaveChanges();
                }
                ts.Complete();
            }
        }
        catch (Exception ex)
        {
            Debug.WriteLine(ex);
        }
        finally
        {
            dTx?.Complete();
            dTx?.Dispose(); // 显式释放DependentTransaction
            countdown?.Signal();
        }
    }
}

3. 改用异步任务替代线程池（推荐） #

使用Task.Run结合async/await模式，更符合现代.NET并发编程模型，同时更好地支持事务流：

[TestMethod]
public async Task SaveTwoDocumentsSimultaneously_WorkSuccessfully()
{
    //Assign
    var doc1 = new Document() {Number = "Test1"};
    var doc2 = new Document() {Number = "Test2"};

    //action
    var task1 = Task.Run(() => SaveDocument(doc1));
    var task2 = Task.Run(() => SaveDocument(doc2));
    await Task.WhenAll(task1, task2);

    //assert
    //assertion code for check two document inserted
    ....
}

private void SaveDocument(Document doc)
{
    using (var dTx = Transaction.Current.DependentClone(DependentCloneOption.BlockCommitUntilComplete))
    using (var ts = new TransactionScope(dTx))
    {
        using (var ctx = new PlanningDbContext())
        {
            ctx.Documents.Add(doc);
            ctx.SaveChanges();
        }
        ts.Complete();
        dTx.Complete();
    }
}

额外注意事项 #

• 若使用SQL Server，确保数据库支持事务并发操作，无需额外开启MSDTC（单实例本地测试默认支持）。
• 断言代码需在countdown.Wait()（或await Task.WhenAll）之后、基类Cleanup之前执行，否则无法读取事务中未提交的数据。

内容的提问来源于stack exchange，提问作者Masoud