Transactionscope issues when using multiple queues/workers

question
1

I am using hangfire to run background jobs and each user in my system have its own queue containing one worker instance. The job that is performed by each worker is wrapped in a TransactionScope. The function that is called by each worker looks like the following.

using (TransactionScope transactionScope = TransactionHandling.GetTransactionScope(60))
{  
  doWork();
  transactionScope.Complete();                
}

Dowork() simply does database update/insert queries and ws-calls, etc. The GetTransactionScope(timeout seconds) creates a transactionscope with isolation level ReadCommitted and TransactionScopeOptions required (GetTransactionScope also supports suppress for testing purposes with SQL CE).

When worker 1 in performing a job from queue A and worker 2 starts performing a job from queue B before worker 1 is finished I’m getting the following exception. Note the workers all run the same code block, regardless of which queue they are serving

System.Data.Entity.Core.EntityException: The underlying provider failed on EnlistTransaction. —> System.InvalidOperationException: Connection currently has transaction enlisted. Finish current transaction and retry.
at System.Data.SqlClient.SqlConnection.EnlistTransaction(Transaction transaction)
at System.Data.Entity.Infrastructure.Interception.InternalDispatcher1.Dispatch[TTarget,TInterceptionContext](TTarget target, Action2 operation, TInterceptionContext interceptionContext, Action3 executing, Action3 executed)…

How can I avoid this exception, I must have missed something in the docs…
Thanks

2

Please show me the code that creates and closes connections to a database.

3

Its failing in a function called GetWorkQueueItems. The function that fails uses my DAL, which implements a repository pattern. The constructor takes a EF db context as created by Autofac

public class CalculationEngineWorkQueueRepository : Repository<DAL.CalculationEngineWorkQueue>, ICalculationEngineWorkQueueRepository
    {
        private readonly KonstruktEntities _context;
        public CalculationEngineWorkQueueRepository(KonstruktEntities context)
            : base(context)
        {
            _context = context;
        }

        public IQueryable<CalculationEngineWorkQueue> GetWorkQueueItems(string userId, int budgetId, long executionId)
        {
            return SearchFor(x => x.UserId == userId &&
                x.IsProcessed == false &&
                x.CalculationLine.BudgetID == budgetId &&
                x.ExecutionId == executionId &&
                x.DestinationEntity == (int)EntityType.UserBudgetLine).OrderBy(x => x.Timestamp);
        }
        public IQueryable<CalculationEngineWorkQueue> GetWorkQueueItems(string userId, int budgetId)
        {
            return SearchFor(x => x.UserId == userId &&
                x.IsProcessed == false &&
                x.CalculationLine.BudgetID == budgetId &&
                x.DestinationEntity == (int)EntityType.UserBudgetLine).OrderBy(x => x.Timestamp);
        }


        public IQueryable<CalculationEngineWorkQueue> GetWorkQueueItems(EntityType entityType, int budgetId)
        {
            return this.SearchFor(x => x.CalculationLine.BudgetID == budgetId &&
                                       x.IsProcessed == false &&
                                       x.DestinationEntity == (int)entityType).OrderBy(x => x.Timestamp);
        }

Here is the base repository class

public interface IRepository<T>
    {
        void Insert(T entity);
        void Delete(T entity);
        IQueryable<T> SearchFor(Expression<Func<T, bool>> predicate);
        IQueryable<T> GetAll();
        T GetById(int id);
    }

    public abstract class Repository<T> : IRepository<T> where T : class
    {
        protected DbSet<T> DbSet;

        public Repository(DbContext dataContext)
        {
            DbSet = dataContext.Set<T>();
        }

        #region IRepository<T> Members

        public void Insert(T entity)
        {
            DbSet.Add(entity);
        }
        
        public void Delete(T entity)
        {
            DbSet.Remove(entity);
        }

        public IQueryable<T> SearchFor(Expression<Func<T, bool>> predicate)
        {
            return DbSet.Where(predicate);
        }

        public IQueryable<T> GetAll()
        {
            return DbSet;
        }

        public T GetById(int id)
        {
            return DbSet.Find(id);
        }

        #endregion
    }

…and finally this is how I instantiate the context using autofac

            builder.Register(c =>
            {
                var context = new KonstruktEntities();
                context.Database.CommandTimeout = 300;
                return context;
            }).AsSelf().InstancePerLifetimeScope();
4

I tried to change the isolation level to ReadUnCommitted (can be done as the workers are not dependant on each other) but still the same problem…

5

Where do you create the lifetime scope? If you don’t create it for each background job, then the global lifetime is used, so your database context instance is used from multiple threads at the same time that gives you the exception above.

6

Hmm it could be the root scope in this case. not sure though… possibly I should use InstancePerRequest instead…