Merge lp:~adeuring/lazr.jobrunner/early-reference-to-job.fail-test into lp:lazr.jobrunner

[Abel Deuring]

This branch adds a test to imitate the possible behavviour of Storm
based job instances: That accessing methods or other of the instance
may fail when a DB error occurred. The test uses the job class
VolatileAttributesJob, where accessing the method 'fail' fails after
run() has been called.

I also noticed that JobRunner.run() left jobs in the status RUNNING,
if the methods suspend(), queue(), complete() fail. This can easily
happen, for example, when the method run() of a Storm based class seems
to succeed, but the call of transaction.commit() in job.complete()
fails due to an IntegrityError.

Hence I reorganised the exception handling in JobRunner.run() a bit,
so that job.fail() is called when suspend(), queue(), complete()
raise an error.

[Aaron Bentley (abentley)]

Good catch.

review: Approve

=== modified file 'src/lazr/jobrunner/jobrunner.py'
--- src/lazr/jobrunner/jobrunner.py	2012-03-26 09:09:17 +0000
+++ src/lazr/jobrunner/jobrunner.py	2012-03-26 12:51:49 +0000
@@ -159,29 +159,30 @@
         do_retry = False
         try:
             try:
-                job.run()
-            except job.retry_error_types, e:
-                if job.attempt_count > job.max_retries:
-                    raise
-                self.logger.exception(
-                    "Scheduling retry due to %s: %s." % (
-                        e.__class__.__name__, e))
-                do_retry = True
-        except SuspendJobException:
-            self.logger.debug("Job suspended itself")
-            job.suspend(manage_transaction=True)
-            self.incomplete_jobs.append(job)
+                try:
+                    job.run()
+                except job.retry_error_types, e:
+                    if job.attempt_count > job.max_retries:
+                        raise
+                    self.logger.exception(
+                        "Scheduling retry due to %s: %s." % (
+                            e.__class__.__name__, e))
+                    do_retry = True
+            except SuspendJobException:
+                self.logger.debug("Job suspended itself")
+                job.suspend(manage_transaction=True)
+                self.incomplete_jobs.append(job)
+            else:
+                if do_retry:
+                    job.queue(manage_transaction=True)
+                    self.incomplete_jobs.append(job)
+                else:
+                    job.complete(manage_transaction=True)
+                    self.completed_jobs.append(job)
         except Exception:
             fail(manage_transaction=True)
             self.incomplete_jobs.append(job)
             raise
-        else:
-            if do_retry:
-                job.queue(manage_transaction=True)
-                self.incomplete_jobs.append(job)
-            else:
-                job.complete(manage_transaction=True)
-                self.completed_jobs.append(job)
 
     def runJobHandleError(self, job):
         """Run the specified job, handling errors."""

=== modified file 'src/lazr/jobrunner/tests/test_jobrunner.py'
--- src/lazr/jobrunner/tests/test_jobrunner.py	2012-03-26 09:09:17 +0000
+++ src/lazr/jobrunner/tests/test_jobrunner.py	2012-03-26 12:51:49 +0000
@@ -36,7 +36,8 @@
 
     retry_error_types = ()
 
-    def __init__(self, job_id, failure=None):
+    def __init__(self, job_id, failure=None, error_in_queue=False,
+                 error_in_suspend=False, error_in_complete=False):
         super(FakeJob, self).__init__(job_id)
         self.unrun = True
         self.failure = failure
@@ -44,6 +45,9 @@
         self.notifyUserError_called = False
         self.queue_call_count = 0
         self.lease_held = False
+        self.error_in_queue = error_in_queue
+        self.error_in_suspend = error_in_suspend
+        self.error_in_complete = error_in_complete
 
     def save(self):
         pass
@@ -63,6 +67,8 @@
         self.save()
 
     def complete(self, manage_transaction=False):
+        if self.error_in_complete:
+            raise Exception('complete() failed')
         super(FakeJob, self).complete(manage_transaction)
         self.save()
 
@@ -74,9 +80,16 @@
         self.notifyOops_called = True
 
     def queue(self, manage_transaction=False):
+        if self.error_in_queue:
+            raise Exception('queue() failed')
         super(FakeJob, self).queue(manage_transaction)
         self.queue_call_count += 1
 
+    def suspend(self, manage_transaction=False):
+        if self.error_in_suspend:
+            raise Exception('suspend() failed')
+        super(FakeJob, self).suspend(manage_transaction)
+
     def notifyUserError(self, exception):
         self.notifyUserError_called = True
 
@@ -84,6 +97,22 @@
         return [('foo', 'bar')]
 
 
+class VolatileAttributesJob:
+
+    def __init__(self, job_id):
+        self.job = FakeJob(job_id)
+        self.failed = False
+
+    def run(self):
+        self.failed = True
+        raise Exception('foo')
+
+    def __getattr__(self, name):
+        if name == 'fail' and self.failed:
+            raise AttributeError('no access')
+        return getattr(self.job, name)
+
+
 class OOPSTestRepository:
     """Record OOPSes in memory."""
 
@@ -257,3 +286,34 @@
             oopsMessage=message_storage.oopsMessage)
         report = runner.runJobHandleError(job)
         self.assertEqual({'foo': 'bar'}, report['extra_data'])
+
+    def test_job_with_volatile_attributes(self):
+        # Properties of jobs that are Storm objects may not be accessible
+        # when a database transaction failed. In this case, even the
+        # method Job.fail() may not be accessible. JobRUnner.run()
+        # handles this by storing an early reference to job.fail().
+        job = VolatileAttributesJob(1)
+        self.runner.runJobHandleError(job)
+        self.assertEqual(JobStatus.FAILED, job.status)
+
+    def test_job_fails_in_complete(self):
+        # If job.complete() fails, job.fail() is called.
+        job = FakeJob(1, error_in_complete=True)
+        self.runner.runJobHandleError(job)
+        self.assertEqual(JobStatus.FAILED, job.status)
+
+    def test_job_fails_in_suspend(self):
+        # If job.suspend() fails, job.fail() is called.
+        job = FakeJob(1, SuspendJobException, error_in_suspend=True)
+        self.runner.runJobHandleError(job)
+        self.assertEqual(JobStatus.FAILED, job.status)
+
+    def test_job_fails_in_queue(self):
+        # If job.queue() fails, job.fail() is called.
+        class TryAgain(Exception):
+            pass
+
+        job = FakeJob(1, TryAgain('once more'), error_in_queue=True)
+        job.retry_error_types = (TryAgain, )
+        self.runner.runJobHandleError(job)
+        self.assertEqual(JobStatus.FAILED, job.status)