Merge lp:~abentley/launchpad/short-celrerybeat-transactions into lp:launchpad

[Aaron Bentley]

run_missing_ready closes transactions and uses its own user

= Summary =
Fix bug #1003720: celerybeat keeps transactions open forever

== Proposed fix ==
Commit transactions and use a unique dbuser

== Pre-implementation notes ==
None

== LOC Rationale ==
I have a LOC credit of 1964.

== Implementation details ==
Can't think of any

== Tests ==
bin/test test_celeryjob

== Demo and Q/A ==
Enable celerybeat. After it has run run_missing_ready once, ask webops whether any transactions are held.

= Launchpad lint =

Checking for conflicts and issues in changed files.

Linting changed files:
  lib/lp/services/job/celeryjob.py
  database/schema/security.cfg
  lib/lp/services/job/tests/test_celeryjob.py

[j.c.sackett (jcsackett)]

Looks good, Aaron. Thanks.

review: Approve

=== modified file 'database/schema/security.cfg'
--- database/schema/security.cfg	2012-05-24 05:43:49 +0000
+++ database/schema/security.cfg	2012-05-24 20:10:35 +0000
@@ -1948,6 +1948,13 @@
 public.job                              = SELECT, UPDATE
 type=user
 
+[run_missing_ready]
+groups=script
+public.branchjob                        = SELECT
+public.job                              = SELECT
+public.featureflag                      = SELECT
+type=user
+
 [updateremoteproduct]
 groups=script
 public.accesspolicy                     = SELECT, INSERT

=== modified file 'lib/lp/services/job/celeryjob.py'
--- lib/lp/services/job/celeryjob.py	2012-05-14 20:41:20 +0000
+++ lib/lp/services/job/celeryjob.py	2012-05-24 20:10:35 +0000
@@ -25,10 +25,8 @@
 from zope.component import getUtility
 
 from lp.code.model.branchjob import BranchScanJob
-from lp.services.config import (
-    config,
-    dbconfig,
-    )
+from lp.scripts.helpers import TransactionFreeOperation
+from lp.services.config import dbconfig
 from lp.services.database.lpstorm import IStore
 from lp.services.features import (
     install_feature_controller,
@@ -89,14 +87,16 @@
     :param _no_init: For tests.  If True, do not perform the initialization.
     """
     if not _no_init:
-        task_init(config.launchpad.dbuser)
-    count = 0
-    for job in find_missing_ready(BranchScanJob):
-        if not celery_enabled(job.__class__.__name__):
-            continue
-        job.celeryCommitHook(True)
-        count += 1
-    info('Scheduled %d missing jobs.', count)
+        task_init('run_missing_ready')
+    with TransactionFreeOperation():
+        count = 0
+        for job in find_missing_ready(BranchScanJob):
+            if not celery_enabled(job.__class__.__name__):
+                continue
+            job.celeryCommitHook(True)
+            count += 1
+        info('Scheduled %d missing jobs.', count)
+        transaction.commit()
 
 
 needs_zcml = True

=== modified file 'lib/lp/services/job/tests/test_celeryjob.py'
--- lib/lp/services/job/tests/test_celeryjob.py	2012-05-14 20:06:58 +0000
+++ lib/lp/services/job/tests/test_celeryjob.py	2012-05-24 20:10:35 +0000
@@ -2,12 +2,14 @@
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 from lp.code.model.branchjob import BranchScanJob
+from lp.scripts.helpers import TransactionFreeOperation
 from lp.services.features.testing import FeatureFixture
 from lp.services.job.tests import (
     drain_celery_queues,
     monitor_celery,
     )
 from lp.testing import TestCaseWithFactory
+from lp.testing.dbuser import dbuser
 from lp.testing.layers import ZopelessAppServerLayer
 
 
@@ -42,7 +44,9 @@
         """run_missing_ready does nothing if the class isn't enabled."""
         self.createMissingJob()
         with monitor_celery() as responses:
-            self.run_missing_ready(_no_init=True)
+            with dbuser('run_missing_ready'):
+                with TransactionFreeOperation.require():
+                    self.run_missing_ready(_no_init=True)
         self.assertEqual([], responses)
 
     def test_run_missing_ready(self):
@@ -51,5 +55,7 @@
         self.useFixture(
             FeatureFixture({'jobs.celery.enabled_classes': 'BranchScanJob'}))
         with monitor_celery() as responses:
-            self.run_missing_ready(_no_init=True)
+            with dbuser('run_missing_ready'):
+                with TransactionFreeOperation.require():
+                    self.run_missing_ready(_no_init=True)
         self.assertEqual(1, len(responses))