Enqueueing a job does not trigger a prepared transaction anymore when using transaction scope enlistment (#160)

* Added test for prepared transactions

* Enqueueing a job does not trigger a prepared transaction anymore when using transaction scope enlistment

Co-authored-by: Johannes Hartmann <[email protected]>

Author: jhartmann123

src/Hangfire.PostgreSql/IPersistentJobQueue.cs

@@ -19,6 +19,7 @@
 //   
 //    Special thanks goes to him.
 
+using System.Data;
 using System.Threading;
 using Hangfire.Storage;
 
@@ -27,6 +28,6 @@ namespace Hangfire.PostgreSql
     public interface IPersistentJobQueue
     {
         IFetchedJob Dequeue(string[] queues, CancellationToken cancellationToken);
-        void Enqueue(string queue, string jobId);
+        void Enqueue(IDbConnection connection, string queue, string jobId);
     }
 }

src/Hangfire.PostgreSql/PostgreSqlJobQueue.cs

@@ -229,14 +229,14 @@ ORDER BY
 				markJobAsFetched.Queue);
 		}
 
-		public void Enqueue(string queue, string jobId)
+		public void Enqueue(IDbConnection connection, string queue, string jobId)
 		{
 			string enqueueJobSql = @"
 INSERT INTO """ + _options.SchemaName + @""".""jobqueue"" (""jobid"", ""queue"") 
 VALUES (@jobId, @queue);
 ";
 
-			_storage.UseConnection(connection => connection.Execute(enqueueJobSql, new {jobId = Convert.ToInt32(jobId, CultureInfo.InvariantCulture), queue = queue}));
+			connection.Execute(enqueueJobSql, new {jobId = Convert.ToInt32(jobId, CultureInfo.InvariantCulture), queue = queue});
 		}
 
 		[UsedImplicitly(ImplicitUseTargetFlags.WithMembers)]

src/Hangfire.PostgreSql/PostgreSqlWriteOnlyTransaction.cs

@@ -169,7 +169,7 @@ public override void AddToQueue(string queue, string jobId)
             var provider = _queueProviders.GetProvider(queue);
             var persistentQueue = provider.GetJobQueue();
 
-            QueueCommand((con) => persistentQueue.Enqueue(queue, jobId));
+            QueueCommand((con) => persistentQueue.Enqueue(con, queue, jobId));
         }
 
         public override void IncrementCounter(string key)

tests/Hangfire.PostgreSql.Tests/PostgreSqlJobQueueFacts.cs

@@ -53,9 +53,9 @@ public void Dequeue_ShouldFetchAJob_FromQueueWithHigherPriority()
                 var queue = CreateJobQueue(storage, false);
                 var token = CreateTimingOutCancellationToken();
 
-                queue.Enqueue("1", "1");
-                queue.Enqueue("2", "2");
-                queue.Enqueue("3", "3");
+                queue.Enqueue(connection, "1", "1");
+                queue.Enqueue(connection, "2", "2");
+                queue.Enqueue(connection, "3", "3");
 
                 Assert.Equal("1", queue.Dequeue(new[] { "1", "2", "3" }, token).JobId);
                 Assert.Equal("2", queue.Dequeue(new[] { "2", "3", "1" }, token).JobId);
@@ -436,7 +436,7 @@ public void Queues_Should_Support_Long_Queue_Names()
 
 				Assert.True(name.Length > 21);
 
-				queue.Enqueue(name, "1");
+				queue.Enqueue(connection, name, "1");
 
 				var record = connection.Query(@"select * from """ + GetSchemaName() + @""".""jobqueue""").Single();
 				Assert.Equal(name, record.queue.ToString());
@@ -449,7 +449,7 @@ private void Enqueue_AddsAJobToTheQueue(bool useNativeDatabaseTransactions)
 			{
 				var queue = CreateJobQueue(storage, useNativeDatabaseTransactions);
 
-				queue.Enqueue("default", "1");
+				queue.Enqueue(connection, "default", "1");
 
 				var record = connection.Query(@"select * from """ + GetSchemaName() + @""".""jobqueue""").Single();
 				Assert.Equal("1", record.jobid.ToString());

tests/Hangfire.PostgreSql.Tests/PostgreSqlWriteOnlyTransactionFacts.cs

@@ -5,6 +5,7 @@
 using System.Linq;
 using System.Transactions;
 using Dapper;
+using Hangfire.Common;
 using Hangfire.States;
 using Moq;
 using Npgsql;
@@ -265,7 +266,7 @@ public void AddToQueue_CallsEnqueue_OnTargetPersistentQueue()
 
                 Commit(sql, x => x.AddToQueue("default", "1"));
 
-                correctJobQueue.Verify(x => x.Enqueue("default", "1"));
+                correctJobQueue.Verify(x => x.Enqueue(sql, "default", "1"));
             });
         }
 
@@ -1051,6 +1052,35 @@ public void PersistList_ClearsExpirationTime_OnAGivenHash()
             });
         }
 
+        [Fact, CleanDatabase]
+        public void AddToQueue_AddsAJobToTheQueue_UsingStorageConnection_WithTransactionScopeEnlistment()
+        {
+            string jobId;
+            var storage = new PostgreSqlStorage(ConnectionUtils.GetConnectionString(), new PostgreSqlStorageOptions { EnableTransactionScopeEnlistment = true });
+            using (var storageConnection = storage.GetConnection())
+            {
+                using (var writeTransaction = storageConnection.CreateWriteTransaction())
+                {
+                    // Explicitly call multiple write commands here, as AddToQueue previously opened an own connection.
+                    // This triggered a prepared transaction which should be avoided.
+                    jobId = storageConnection.CreateExpiredJob(Job.FromExpression(() => Console.Write("Hi")), new Dictionary<string, string>(), DateTime.UtcNow, TimeSpan.FromMinutes(1));
+
+                    writeTransaction.SetJobState(jobId, new ScheduledState(DateTime.UtcNow));
+                    writeTransaction.AddToQueue("default", jobId);
+                    writeTransaction.PersistJob(jobId);
+                    writeTransaction.Commit();
+                }
+            }
+
+            UseConnection(connection =>
+            {
+                var record = connection.Query(@"select * from """ + GetSchemaName() + @""".""jobqueue""").Single();
+                Assert.Equal(jobId, record.jobid.ToString());
+                Assert.Equal("default", record.queue);
+                Assert.Null(record.FetchedAt);
+            });
+        }
+
         private void UseConnection(Action<NpgsqlConnection> action)
         {
             using (var connection = ConnectionUtils.CreateConnection())