Merge pull request #682 from Particular/adopt_batches

AdoptOrphanedBatches running more then necessary

Author: johnsimons

src/ServiceControl/Infrastructure/TimeKeeper.cs

@@ -10,21 +10,23 @@ public class TimeKeeper : IDisposable
         ConcurrentDictionary<Timer, object> timers = new ConcurrentDictionary<Timer, object>();
         private ILog log = LogManager.GetLogger<TimeKeeper>();
 
-        public Timer New(Action callback, TimeSpan dueTime, TimeSpan period)
+        public Timer NewTimer(Func<bool> callback, TimeSpan dueTime, TimeSpan period)
         {
             Timer timer = null;
 
             timer = new Timer(_ =>
             {
+                var reschedule = false;
+
                 try
                 {
-                    callback();
+                    reschedule = callback();
                 }
                 catch (Exception ex)
                 {
                     log.Error("Reoccurring timer task failed.", ex);
                 }
-                if (timers.ContainsKey(timer))
+                if (reschedule && timers.ContainsKey(timer))
                 {
                     try
                     {
@@ -41,6 +43,15 @@ public Timer New(Action callback, TimeSpan dueTime, TimeSpan period)
             return timer;
         }
 
+        public Timer New(Action callback, TimeSpan dueTime, TimeSpan period)
+        {
+            return NewTimer(() =>
+            {
+                callback();
+                return true;
+            }, dueTime, period);
+        }
+
         public void Release(Timer timer)
         {
             object _;

src/ServiceControl/Recoverability/Retries/FailedMessageRetries.cs

@@ -76,20 +76,23 @@ public AdoptOrphanBatchesFromPreviousSession(RetryDocumentManager retryDocumentM
                 this.timeKeeper = timeKeeper;
             }
 
-            private void AdoptOrphanedBatches()
+            private bool AdoptOrphanedBatches()
             {
-                var allDone = retryDocumentManager.AdoptOrphanedBatches();
+                bool hasMoreWorkToDo;
+                retryDocumentManager.AdoptOrphanedBatches(out hasMoreWorkToDo);
 
-                if (allDone)
+                if (!hasMoreWorkToDo)
                 {
                     //Disable timeout
                     timer.Change(Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan);
                 }
+
+                return hasMoreWorkToDo;
             }
 
             protected override void OnStart()
             {
-                timer = timeKeeper.New(AdoptOrphanedBatches, TimeSpan.Zero, TimeSpan.FromMinutes(2));
+                timer = timeKeeper.NewTimer(AdoptOrphanedBatches, TimeSpan.Zero, TimeSpan.FromMinutes(2));
             }
 
             protected override void OnStop()

src/ServiceControl/Recoverability/Retries/RetryDocumentManager.cs

@@ -107,7 +107,7 @@ public void RemoveFailedMessageRetryDocument(string uniqueMessageId)
             Store.DatabaseCommands.Delete(FailedMessageRetry.MakeDocumentId(uniqueMessageId), null);
         }
 
-        internal bool AdoptOrphanedBatches()
+        internal void AdoptOrphanedBatches(out bool hasMoreWorkToDo)
         {
             using (var session = Store.OpenSession())
             {
@@ -123,8 +123,13 @@ internal bool AdoptOrphanedBatches()
 
                 AdoptBatches(session, orphanedBatchIds);
 
-                var moreToDo = stats.IsStale || orphanedBatchIds.Any();
-                return abort || !moreToDo;
+                if (abort)
+                {
+                    hasMoreWorkToDo = false;
+                    return;
+                }
+
+                hasMoreWorkToDo = stats.IsStale || orphanedBatchIds.Any();
             }
         }