Added locking to the DuplicateCheckerPlugin as there were race conditions

Author: niemyjski

Source/Shared/Plugins/Default/1010_DuplicateCheckerPlugin.cs

@@ -10,6 +10,7 @@ namespace Exceptionless.Plugins.Default {
     public class DuplicateCheckerPlugin : IEventPlugin, IDisposable {
         private readonly ConcurrentQueue<Tuple<int, DateTimeOffset>> _processed = new ConcurrentQueue<Tuple<int, DateTimeOffset>>();
         private readonly ConcurrentQueue<MergedEvent> _mergedEvents = new ConcurrentQueue<MergedEvent>();
+        private readonly object _lock = new object();
         private readonly TimeSpan _interval;
         private Timer _timer;
 
@@ -31,40 +32,47 @@ public void Run(EventPluginContext context) {
             int hashCode = context.Event.GetHashCode();
             int count = context.Event.Count ?? 1;
             context.Log.FormattedTrace(typeof(DuplicateCheckerPlugin), String.Concat("Checking event: ", context.Event.Message, " with hash: ", hashCode));
+            
+            lock (_lock) {
+                context.Log.Debug("Got Lock on thread" + Thread.CurrentThread.ManagedThreadId);
 
-            // Increment the occurrence count if the event is already queued for submission.
-            var merged = _mergedEvents.FirstOrDefault(s => s.HashCode == hashCode);
-            if (merged != null) {
-                merged.IncrementCount(count);
-                context.Log.FormattedInfo(typeof(DuplicateCheckerPlugin), String.Concat("Ignoring duplicate event with hash:", hashCode));
-                context.Cancel = true;
-                return;
-            }
+                // Increment the occurrence count if the event is already queued for submission.
+                var merged = _mergedEvents.FirstOrDefault(s => s.HashCode == hashCode);
+                if (merged != null) {
+                    merged.IncrementCount(count);
+                    context.Log.FormattedInfo(typeof(DuplicateCheckerPlugin), String.Concat("Ignoring duplicate event with hash:", hashCode));
+                    context.Cancel = true;
+                    return;
+                }
+
+                DateTimeOffset repeatWindow = DateTimeOffset.UtcNow.Subtract(_interval);
+                if (_processed.Any(s => s.Item1 == hashCode && s.Item2 >= repeatWindow)) {
+                    context.Log.FormattedInfo(typeof(DuplicateCheckerPlugin), String.Concat("Adding event with hash:", hashCode, " to cache."));
+                    // This event is a duplicate for the first time, lets save it so we can delay it while keeping count
+                    _mergedEvents.Enqueue(new MergedEvent(hashCode, context, count));
+                    context.Cancel = true;
+                    return;
+                }
 
-            DateTimeOffset repeatWindow = DateTimeOffset.UtcNow.Subtract(_interval);
-            if (_processed.Any(s => s.Item1 == hashCode && s.Item2 >= repeatWindow)) {
-                context.Log.FormattedInfo(typeof(DuplicateCheckerPlugin), String.Concat("Adding event with hash:", hashCode, " to cache."));
-                // This event is a duplicate for the first time, lets save it so we can delay it while keeping count
-                _mergedEvents.Enqueue(new MergedEvent(hashCode, context, count));
-                context.Cancel = true;
-            } else {
                 context.Log.FormattedInfo(typeof(DuplicateCheckerPlugin), String.Concat("Enqueueing event with hash:", hashCode, " to cache."));
                 _processed.Enqueue(Tuple.Create(hashCode, DateTimeOffset.UtcNow));
+
+                Tuple<int, DateTimeOffset> temp;
+                while (_processed.Count > 50)
+                    _processed.TryDequeue(out temp);
             }
-            
-            Tuple<int, DateTimeOffset> temp;
-            while (_processed.Count > 50)
-                _processed.TryDequeue(out temp);
         }
 
         private void OnTimer(object state) {
             EnqueueMergedEvents();
         }
 
         private void EnqueueMergedEvents() {
-            MergedEvent mergedEvent;
-            while (_mergedEvents.TryDequeue(out mergedEvent))
-                mergedEvent.Enqueue();
+            lock (_lock) {
+                MergedEvent mergedEvent;
+                while (_mergedEvents.TryDequeue(out mergedEvent))
+                    mergedEvent.Enqueue();
+            }
         }
 
         public void Dispose() {

Source/Tests/Plugins/PluginTests.cs

@@ -575,18 +575,21 @@ public void VerifyDeduplicationMultithreaded() {
             var errorPlugin = new ErrorPlugin();
 
             List<EventPluginContext> contexts = new List<EventPluginContext>();
-            using (var duplicateCheckerPlugin = new DuplicateCheckerPlugin(TimeSpan.FromMilliseconds(20))) {
-                Parallel.For(0, 10, index => {
+            using (var duplicateCheckerPlugin = new DuplicateCheckerPlugin(TimeSpan.FromMilliseconds(100))) {
+                var result = Parallel.For(0, 10, index => {
                     var builder = GetException().ToExceptionless();
                     var context = new EventPluginContext(client, builder.Target, builder.PluginContextData);
                     contexts.Add(context);
 
                     errorPlugin.Run(context);
                     duplicateCheckerPlugin.Run(context);
                 });
+
+                while (!result.IsCompleted)
+                    Thread.Sleep(1);
             }
 
-            Thread.Sleep(50);
+            Thread.Sleep(150);
             Assert.Equal(1, contexts.Count(c => !c.Cancel));
             Assert.Equal(9, contexts.Count(c => c.Cancel));
             Assert.Equal(9, contexts.Sum(c => c.Event.Count.GetValueOrDefault()));