[CalendarServer-changes] [12893] twext/trunk/twext/enterprise

[source_changes at macosforge.org]

Revision: 12893
          http://trac.calendarserver.org//changeset/12893
Author:   cdaboo at apple.com
Date:     2014-03-13 11:21:52 -0700 (Thu, 13 Mar 2014)
Log Message:
-----------
Make sure lost periodic work sweep keeps going if one of the lost jobs raises an exception.

Modified Paths:
--------------
    twext/trunk/twext/enterprise/jobqueue.py
    twext/trunk/twext/enterprise/test/test_jobqueue.py

Modified: twext/trunk/twext/enterprise/jobqueue.py
===================================================================
--- twext/trunk/twext/enterprise/jobqueue.py	2014-03-13 18:09:35 UTC (rev 12892)
+++ twext/trunk/twext/enterprise/jobqueue.py	2014-03-13 18:21:52 UTC (rev 12893)
@@ -1385,7 +1385,10 @@
             )
             for overdueItem in overdueItems:
                 peer = self.choosePerformer()
-                yield peer.performJob(overdueItem.jobID)
+                try:
+                    yield peer.performJob(overdueItem.jobID)
+                except Exception as e:
+                    log.err("Failed to perform periodic lost job for jobid={}, {}".format(overdueItem.jobID, e))
 
         return inTransaction(self.transactionFactory, workCheck)
 

Modified: twext/trunk/twext/enterprise/test/test_jobqueue.py
===================================================================
--- twext/trunk/twext/enterprise/test/test_jobqueue.py	2014-03-13 18:09:35 UTC (rev 12892)
+++ twext/trunk/twext/enterprise/test/test_jobqueue.py	2014-03-13 18:21:52 UTC (rev 12893)
@@ -233,6 +233,8 @@
     """
 
     def doWork(self):
+        if self.a == -1:
+            raise ValueError("Ooops")
         return DummyWorkDone.makeJob(
             self.transaction, jobID=self.jobID + 100, workID=self.workID + 100, aPlusB=self.a + self.b
         )
@@ -737,7 +739,54 @@
         )
 
 
+    @inlineCallbacks
+    def test_exceptionWhenCheckingForLostWork(self):
+        """
+        L{PeerConnectionPool._periodicLostWorkCheck} should execute any
+        outstanding work items, and keep going if some raise an exception.
+        """
+        dbpool = buildConnectionPool(self, nodeSchema + jobSchema + schemaText)
+        # An arbitrary point in time.
+        fakeNow = datetime.datetime(2012, 12, 12, 12, 12, 12)
+        # *why* does datetime still not have .astimestamp()
+        sinceEpoch = astimestamp(fakeNow)
+        clock = Clock()
+        clock.advance(sinceEpoch)
+        qpool = PeerConnectionPool(clock, dbpool.connection, 0)
 
+        # Let's create a couple of work items directly, not via the enqueue
+        # method, so that they exist but nobody will try to immediately execute
+        # them.
+
+        @transactionally(dbpool.connection)
+        @inlineCallbacks
+        def setup(txn):
+            # First, one that's right now.
+            yield DummyWorkItem.makeJob(
+                txn, a=1, b=0, notBefore=fakeNow - datetime.timedelta(20 * 60)
+            )
+
+            # Next, create one that's actually far enough into the past to run.
+            yield DummyWorkItem.makeJob(
+                txn, a=-1, b=1, notBefore=fakeNow - datetime.timedelta(20 * 60)
+            )
+
+            # Finally, one that's actually scheduled for the future.
+            yield DummyWorkItem.makeJob(
+                txn, a=2, b=0, notBefore=fakeNow - datetime.timedelta(20 * 60)
+            )
+        yield setup
+        yield qpool._periodicLostWorkCheck()
+
+        @transactionally(dbpool.connection)
+        def check(txn):
+            return DummyWorkDone.all(txn)
+
+        every = yield check
+        self.assertEquals([x.aPlusB for x in every], [1, 2])
+
+
+
 class HalfConnection(object):
     def __init__(self, protocol):
         self.protocol = protocol
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <https://lists.macosforge.org/pipermail/calendarserver-changes/attachments/20140313/eb13e215/attachment.html>