Merge lp:~adeuring/launchpad/abort-transaction-in-job-queue into lp:launchpad

[Abel Deuring]

Parameter abort_transaction added to Job.queue()

This branch adds a parameter "abort_transaction" to the method Job.queue().

Jobs are now controlled by the lazr.jobrunner mudule, where I recently
added a timeout mechanism: This Celery based job runner can define
more than one queues for the same jobs, with different timeout values.
If a job times out in a "fast" queue, it cen be re-queue in another
queue with a longer timeout value.

This requires to change the status of the job back to JobStatus.WAITING.
This is done in Job.queue() -- but this method is also called when
a job raises an exception that is listed in Job.retry_error_types.

This retry mechanism assumes that the job left the database in a
consistent state when it raised the "retry exception", hence Job.queue()
calls transaction.commit(). This is mostly likely bad when the job
is interrupted by a timeout, so I added the parameter abort_transaction,
which is True by default.

I also noticed that the paremeter manage_transaction of the methods
Job.start(), Job.complete(), Job.fail() etc was not tested, so I added
these tests:

./bin/test -vvt lp.services.job.tests.test_job.TestJob.test_.*_manages_transactions

no lint

[Abel Deuring (adeuring)]

...and I noticed that the methods start(), complete(),fail() etc of the class IJob were not up to date: I added the parameter manage_transaction.

[Richard Harding (rharding)]

Abel, feel free to correct me here.

- I worry that having method params for manage_transaction that they could be called and not match. So start has it set to true, but not for fail and so we end up with left over transactions not getting processed correctly. It would seem an object level attribute of 'has' or 'uses' transaction would be a better single point to hold that all of the process methods check might help keep things more consistant.
- There's no mention of the LoC policy. Please note in the MP if this is offset somewhere.
- The comments in the test code seem prematurely new lined. For instance: #74-75 would fit in one line per out lint policy. Since this would reduce the LoC impact could these be adjusted?

Thanks for adding the new tests!

review: Needs Information (code*)

[Abel Deuring (adeuring)]

Rick, right, the parameter manage_transaction is indeed good candidate for inconsistencies. The problem here: The job runner implementation in the main LP code (now obsolete) called commit() or abort() where needed. The new implementation in lazr.jobrunner does not do this: It theory. it should completely ignore transactions, because these are relevant for _job_ but not the job runner. The problem with this approach is that some of the methods start(), queue() (sorry, can't remember right now which exactly) are called in parts of the main LP code which is not directly related to "real" job management. (I think there is for example code which creates a job instance and then immediately puts it into the status "suspended".) Committing transactions there would be bad, so I added the parameter manage_transaction. It is only set to True in lazr.jobrunner, but it is used the consistently.

Your suggestion to use something like Job.transaction to automatically decide if start(), complete() etc is very interesting but will require some more thoughts where this property should be added. Another option would be to replace the calls of queue() or suspend() outside the job runner with some other methods like "tweakJobStatusWithoutDBCommits()" -- after all, the job status is supposed to be managed by a job runner but not in model code. (example: lp/buildmaster/model/packagebuild.py, line 278) The we could drop the parameter manage_transaction completely.

But I would like to defer this to another branch.

LoC policy: Once lazr.jobrunner is fully implenented and running, Aaron and I expect that we can remove a reasonable amount of lines from LP.

[Richard Harding (rharding)]

Thanks for the response Abel, ok'ing.

review: Approve (code*)

[j.c.sackett (jcsackett)]

I have nothing further to add to Rick's comments; thanks for this code and the other jobrunner work.

review: Approve

=== modified file 'lib/lp/services/job/interfaces/job.py'
--- lib/lp/services/job/interfaces/job.py	2012-04-06 17:28:25 +0000
+++ lib/lp/services/job/interfaces/job.py	2012-04-12 17:19:19 +0000
@@ -115,22 +115,22 @@
     def getTimeout():
         """Determine how long this job can run before timing out."""
 
-    def start():
+    def start(manage_transaction=False):
         """Mark the job as started."""
 
-    def complete():
+    def complete(manage_transaction=False):
         """Mark the job as completed."""
 
-    def fail():
+    def fail(manage_transaction=False):
         """Indicate that the job has failed permanently.
 
         Only running jobs can fail.
         """
 
-    def queue():
+    def queue(manage_transaction=False, abort_transaction=False):
         """Mark the job as queued for processing."""
 
-    def suspend():
+    def suspend(manage_transaction=False):
         """Mark the job as suspended.
 
         Only waiting jobs can be suspended."""

=== modified file 'lib/lp/services/job/model/job.py'
--- lib/lp/services/job/model/job.py	2012-04-09 19:02:25 +0000
+++ lib/lp/services/job/model/job.py	2012-04-12 17:19:19 +0000
@@ -189,10 +189,12 @@
         if manage_transaction:
             transaction.commit()
 
-    def queue(self, manage_transaction=False):
+    def queue(self, manage_transaction=False, abort_transaction=False):
         """See `IJob`."""
-        # Commit the transaction to update the DB time.
         if manage_transaction:
+            if abort_transaction:
+                transaction.abort()
+            # Commit the transaction to update the DB time.
             transaction.commit()
         self._set_status(JobStatus.WAITING)
         self.date_finished = datetime.datetime.now(UTC)

=== modified file 'lib/lp/services/job/tests/test_job.py'
--- lib/lp/services/job/tests/test_job.py	2012-03-21 16:16:08 +0000
+++ lib/lp/services/job/tests/test_job.py	2012-04-12 17:19:19 +0000
@@ -9,6 +9,7 @@
 import pytz
 from lazr.jobrunner.jobrunner import LeaseHeld
 from storm.locals import Store
+import transaction
 
 from lp.services.database.constants import UTC_NOW
 from lp.services.database.lpstorm import IStore
@@ -252,6 +253,114 @@
             self.assertEqual(
                 status in Job.PENDING_STATUSES, job.is_pending)
 
+    def test_start_manages_transactions(self):
+        # Job.start() does not commit the transaction by default.
+        with TransactionRecorder() as recorder:
+            job = Job()
+            job.start()
+            self.assertEqual([], recorder.transaction_calls)
+
+        # If explicitly specified, Job.start() commits the transaction.
+        with TransactionRecorder() as recorder:
+            job = Job()
+            job.start(manage_transaction=True)
+            self.assertEqual(['commit'], recorder.transaction_calls)
+
+    def test_complete_manages_transactions(self):
+        # Job.complete() does not commit the transaction by default.
+        job = Job()
+        job.start()
+        with TransactionRecorder() as recorder:
+            job.complete()
+            self.assertEqual([], recorder.transaction_calls)
+
+        # If explicitly specified, Job.complete() commits the transaction.
+        job = Job()
+        job.start()
+        with TransactionRecorder() as recorder:
+            job.complete(manage_transaction=True)
+            self.assertEqual(['commit', 'commit'], recorder.transaction_calls)
+
+    def test_fail_manages_transactions(self):
+        # Job.fail() does not commit the transaction by default.
+        job = Job()
+        job.start()
+        with TransactionRecorder() as recorder:
+            job.fail()
+            self.assertEqual([], recorder.transaction_calls)
+
+        # If explicitly specified, Job.fail() commits the transaction.
+        # Note that there is an additional commit to update the job status.
+        job = Job()
+        job.start()
+        with TransactionRecorder() as recorder:
+            job.fail(manage_transaction=True)
+            self.assertEqual(['abort', 'commit'], recorder.transaction_calls)
+
+    def test_queue_manages_transactions(self):
+        # Job.queue() does not commit the transaction by default.
+        job = Job()
+        job.start()
+        with TransactionRecorder() as recorder:
+            job.queue()
+            self.assertEqual([], recorder.transaction_calls)
+
+        # If explicitly specified, Job.queue() commits the transaction.
+        # Note that there is an additional commit to update the job status.
+        job = Job()
+        job.start()
+        with TransactionRecorder() as recorder:
+            job.queue(manage_transaction=True)
+            self.assertEqual(['commit', 'commit'], recorder.transaction_calls)
+
+        # If abort_transaction=True is also passed to Job.queue()
+        # the transaction is first aborted, then two times committed.
+        job = Job()
+        job.start()
+        with TransactionRecorder() as recorder:
+            job.queue(manage_transaction=True, abort_transaction=True)
+            self.assertEqual(
+                ['abort', 'commit', 'commit'], recorder.transaction_calls)
+
+    def test_suspend_manages_transactions(self):
+        # Job.suspend() does not commit the transaction by default.
+        job = Job()
+        job.start()
+        with TransactionRecorder() as recorder:
+            job.suspend()
+            self.assertEqual([], recorder.transaction_calls)
+
+        # If explicitly specified, Job.suspend() commits the transaction.
+        job = Job()
+        job.start()
+        with TransactionRecorder() as recorder:
+            job.suspend(manage_transaction=True)
+            self.assertEqual(['commit'], recorder.transaction_calls)
+
+
+class TransactionRecorder:
+    def __init__(self):
+        self.transaction_calls = []
+
+    def __enter__(self):
+        self.real_commit = transaction.commit
+        self.real_abort = transaction.abort
+        transaction.commit = self.commit
+        transaction.abort = self.abort
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        transaction.commit = self.real_commit
+        transaction.abort = self.real_abort
+
+    def commit(self):
+        self.transaction_calls.append('commit')
+        self.real_commit()
+
+    def abort(self):
+        self.transaction_calls.append('abort')
+        self.real_abort()
+
 
 class TestReadiness(TestCase):
     """Test the implementation of readiness."""