fixed potential race condition when preparing an item for dequeue

Author: Aaron Boxer

src/ServiceStack.Redis/Support/Queue/ISequentialWorkQueue.cs

@@ -17,7 +17,7 @@ public interface ISequentialWorkQueue<T> : IDisposable where T : class
         /// <summary>
         /// Preprare next work item id for dequeueing
         /// </summary>
-        void PrepareNextWorkItem();
+        bool PrepareNextWorkItem();
 
 
         /// <summary>

src/ServiceStack.Redis/Support/Queue/Implementation/RedisSequentialWorkQueue.cs

@@ -68,7 +68,7 @@ public void Enqueue(string workItemId, T workItem)
         /// <summary>
         /// Must call this periodically to move work items from priority queue to pending queue
         /// </summary>
-        public void PrepareNextWorkItem()
+        public bool PrepareNextWorkItem()
         {
             //harvest zombies every 5 minutes
             var now = DateTime.UtcNow;
@@ -86,24 +86,39 @@ public void PrepareNextWorkItem()
                 //1. get next workItemId, or return if there isn't one
                 var smallest = client.ZRangeWithScores(workItemIdPriorityQueue, 0, 0);
                 if (smallest == null || smallest.Length <= 1 ||
-                    RedisNativeClient.ParseDouble(smallest[1]) == CONVENIENTLY_SIZED_FLOAT) return;
+                    RedisNativeClient.ParseDouble(smallest[1]) == CONVENIENTLY_SIZED_FLOAT) return false;
                 var workItemId = client.Deserialize(smallest[0]) as string;
-                using (var pipe = client.CreatePipeline())
+
+                // lock on work item id
+                var lockKey = queueNamespace.GlobalLockKey(workItemId);
+                using (var disposableLock = new DisposableDistributedLock(client, lockKey, lockAcquisitionTimeout, lockTimeout))
                 {
-                    var rawWorkItemId = client.Serialize(workItemId);
+                    // if another client has queued this work item id, then the work item id score
+                    // will be set to CONVENIENTLY_SIZED_FLOAT
+                    var score = client.ZScore(workItemIdPriorityQueue, smallest[0]);
+                    if (score == CONVENIENTLY_SIZED_FLOAT) return false;
 
-                    // lock work item id
-                    pipe.QueueCommand(r => ((RedisNativeClient)r).ZAdd(workItemIdPriorityQueue, CONVENIENTLY_SIZED_FLOAT, smallest[0]));
+                    using (var pipe = client.CreatePipeline())
+                    {
+                        var rawWorkItemId = client.Serialize(workItemId);
 
-                    // track dequeue lock id
-                    pipe.QueueCommand(r => ((RedisNativeClient)r).SAdd(dequeueIdSet, rawWorkItemId));
+                        // lock work item id
+                        pipe.QueueCommand(
+                            r =>
+                            ((RedisNativeClient) r).ZAdd(workItemIdPriorityQueue, CONVENIENTLY_SIZED_FLOAT, smallest[0]));
 
-                    // push into pending set
-                    pipe.QueueCommand(r => ((RedisNativeClient)r).SAdd(pendingWorkItemIdQueue, rawWorkItemId));
+                        // track dequeue lock id
+                        pipe.QueueCommand(r => ((RedisNativeClient) r).SAdd(dequeueIdSet, rawWorkItemId));
+
+                        // push into pending set
+                        pipe.QueueCommand(r => ((RedisNativeClient) r).SAdd(pendingWorkItemIdQueue, rawWorkItemId));
+
+                        pipe.Flush();
+                    }
 
-                    pipe.Flush();
                 }
             }
+            return true;
         }
 
         public ISequentialData<T> Dequeue(int maxBatchSize)

tests/ServiceStack.Redis.Tests/QueueTests.cs

@@ -9,24 +9,32 @@ namespace ServiceStack.Redis.Tests
     [TestFixture]
     public class QueueTests : RedisClientTestsBase
     {
+        const int numMessages = 6;
+        private IList<string> messages0 = new List<string>();
+        private IList<string> messages1 = new List<string>();
+        private string[] patients = new[] {"patient0", "patient1"};
+
+        [SetUp]
+        public override void OnBeforeEachTest()
+        {
+           base.OnBeforeEachTest();
+           for (int i = 0; i < numMessages; ++i)
+           {
+               messages0.Add(String.Format("{0}_message{1}", patients[0], i));
+               messages1.Add(String.Format("{0}_message{1}", patients[1], i));
+           }
+
+        }
 
         [Test]
         public void TestSequentialWorkQueue()
         {
             using (var queue = new RedisSequentialWorkQueue<string>(10,10,"127.0.0.1",6379,1))
             {
-                const int numMessages = 6;
-                var messages0 = new List<string>();
-                var messages1 = new List<string>();
-                var patients = new string[2];
-                patients[0] = "patient0";
-                patients[1] = "patient1";
-
+              
                 for (int i = 0; i < numMessages; ++i)
                 {
-                    messages0.Add(String.Format("{0}_message{1}", patients[0], i));
                     queue.Enqueue(patients[0], messages0[i]);
-                    messages1.Add(String.Format("{0}_message{1}", patients[1], i));
                     queue.Enqueue(patients[1], messages1[i]);
                 }