Merge lp:~allenap/launchpad/multithreaded-checkwatches into lp:launchpad

[Gavin Panella]

[Gavin Panella (allenap)]

This branch makes checkwatches update bug trackers in multiple threads. The
watches for a specific bug tracker are updated sequentially in a single
thread, but other bug trackers may be being updated concurrently in separate
threads.

Most of the changes are refactoring of or new methods for BugWatchUpdater in
checkwatches.py:

 * The existing updateBugTracker() method was renamed to _updateBugTracker().

 * The existing exception handling and oops reporting code in
   updateBugTrackers() was moved to a new updateBugTracker() method.

 * forceUpdateAll() had some identical exception handling code which was
   replaced by a call to the new updateBugTracker() method.

 * updateBugTrackers() was eviscerated and replaced with the thread setup and
   run. It calls a new _bugTrackerUpdaters() method which generates functions
   for the threads to run, each of which will update one bug tracker. These
   functions are put into a work queue from which the threads will pull.

 * A new _interactionDecorator() method was created to support running jobs in
   an interaction. Each of the functions that _bugTrackerUpdaters() yields is
   decorated with this. This is needed because interactions are specific to a
   thread.

Other work:

 * Add a --jobs option to the checkwatches.py script.

 * Tests that called updateBugTracker() now call _updateBugTracker() to avoid
   the transaction stuff.

 * Demonstrate updateBugTrackers() spawning multiple threads.

Testing:

  bin/test -vvct 'checkwatch|externalbug'

Lint:

  lib/lp/bugs/scripts/checkwatches.py
      22: [F0401] Unable to import 'lazr.lifecycle.event' (No module named
          lifecycle)

[Abel Deuring (adeuring)]

Hi Gavin,

great work! And thanks for answering all my somewhat paranoid questions on IRC

review: Approve (code)

=== modified file 'lib/lp/bugs/doc/checkwatches-cli-switches.txt'
--- lib/lp/bugs/doc/checkwatches-cli-switches.txt	2009-10-09 20:52:12 +0000
+++ lib/lp/bugs/doc/checkwatches-cli-switches.txt	2009-11-27 16:20:37 +0000
@@ -138,3 +138,6 @@
       --reset               Update all the watches on the bug tracker,
                             regardless of whether or not they need
                             checking.
+      --jobs=JOBS           The number of simulataneous jobs to run, 1
+                            by default.
+    <BLANKLINE>

=== modified file 'lib/lp/bugs/doc/checkwatches.txt'
--- lib/lp/bugs/doc/checkwatches.txt	2009-11-19 14:41:14 +0000
+++ lib/lp/bugs/doc/checkwatches.txt	2009-11-27 16:20:37 +0000
@@ -46,6 +46,7 @@
     >>> print err
     INFO    creating lockfile
     DEBUG   Using a global batch size of None
+    DEBUG   Skipping updating Ubuntu Bugzilla watches.
     DEBUG   No watches to update on http://bugs.debian.org
     DEBUG   No watches to update on mailto:[e-mail]
     WARNING ExternalBugtracker for BugTrackerType 'SAVANE' is not known.
@@ -53,7 +54,6 @@
     DEBUG   No watches to update on http://sourceforge.net/
     DEBUG   No watches to update on http://bugzilla.gnome.org/
     DEBUG   No watches to update on https://bugzilla.mozilla.org/
-    DEBUG   Skipping updating Ubuntu Bugzilla watches.
     INFO    Time for this run: ... seconds.
     DEBUG   Removing lock file:...
     <BLANKLINE>
@@ -201,7 +201,7 @@
     ...         broken_get_external_bugtracker)
     ...     updater = BugWatchUpdater(transaction)
     ...     updater._login()
-    ...     updater.updateBugTracker(example_bug_tracker)
+    ...     updater._updateBugTracker(example_bug_tracker)
     ... finally:
     ...     externalbugtracker.get_external_bugtracker = (
     ...         real_get_external_bugtracker)
@@ -223,7 +223,7 @@
 a given ExternalBugTracker in each checkwatches run: the batch size.
 
 We need to add some bug watches again since
-BugWatchUpdate.updateBugTracker() automatically rolls back the
+BugWatchUpdate._updateBugTracker() automatically rolls back the
 transaction if something goes wrong.
 
     >>> login('[e-mail]')
@@ -353,7 +353,7 @@
 
     >>> class NonConnectingUpdater(BugWatchUpdater):
     ...
-    ...     def updateBugTracker(self, bug_tracker, batch_size):
+    ...     def _updateBugTracker(self, bug_tracker, batch_size):
     ...         # Update as many watches as the batch size says.
     ...         watches_to_update = (
     ...             bug_tracker.getBugWatchesNeedingUpdate(23)[:batch_size])

=== modified file 'lib/lp/bugs/doc/externalbugtracker.txt'
--- lib/lp/bugs/doc/externalbugtracker.txt	2009-09-22 15:22:53 +0000
+++ lib/lp/bugs/doc/externalbugtracker.txt	2009-11-27 16:20:37 +0000
@@ -1079,7 +1079,8 @@
 
     >>> bug_watch_updater = NonConnectingBugWatchUpdater(
     ...     transaction, QuietFakeLogger())
-    >>> bug_watch_updater.updateBugTracker(standard_bugzilla, batch_size=2)
+    >>> bug_watch_updater._updateBugTracker(
+    ...     standard_bugzilla, batch_size=2)
     initializeRemoteBugDB() called: [u'5', u'6']
     getRemoteStatus() called: u'5'
     getRemoteStatus() called: u'6'
@@ -1089,6 +1090,12 @@
 allows it to be passed as a command-line option when the checkwatches script
 is run.
 
+Before going further, we must abort the current transaction to avoid
+deadlock; updateBugTrackers() runs updateBugTracker() in a different
+thread.
+
+    >>> transaction.abort()
+
     >>> from canonical.launchpad.scripts.logger import FakeLogger
     >>> bug_watch_updater = NonConnectingBugWatchUpdater(
     ...     transaction, FakeLogger())
@@ -1099,3 +1106,92 @@
     initializeRemoteBugDB() called: [u'5', u'6']
     getRemoteStatus() called: u'5'
     getRemoteStatus() called: u'6'
+
+    >>> # We should log in again because updateBugTrackers() logs out.
+    >>> login('[e-mail]')
+
+By default, the updateBugTrackers() only spawns one thread, but it can
+spawn as many as required.
+
+    >>> import threading
+
+    >>> class OutputFileForThreads:
+    ...     def __init__(self):
+    ...         self.output = {}
+    ...         self.lock = threading.Lock()
+    ...     def write(self, data):
+    ...         thread_id = id(threading.currentThread())
+    ...         self.lock.acquire()
+    ...         try:
+    ...             if thread_id in self.output:
+    ...                 self.output[thread_id].append(data)
+    ...             else:
+    ...                 self.output[thread_id] = [data]
+    ...         finally:
+    ...             self.lock.release()
+
+    >>> output_file = OutputFileForThreads()
+
+    >>> class ExternalBugTrackerForThreads(TestExternalBugTracker):
+    ...     def getModifiedRemoteBugs(self, remote_bug_ids, last_checked):
+    ...         print >> output_file, (
+    ...             "getModifiedRemoteBugs(\n"
+    ...             "    remote_bug_ids=%r,\n"
+    ...             "    last_checked=%r)" % (remote_bug_ids, last_checked))
+    ...         return [remote_bug_ids[0], remote_bug_ids[-1]]
+    ...     def getRemoteStatus(self, bug_id):
+    ...         print >> output_file, (
+    ...             "getRemoteStatus(bug_id=%r)" % bug_id)
+    ...         return 'UNKNOWN'
+    ...     def getCurrentDBTime(self):
+    ...         return None
+
+    >>> class BugWatchUpdaterForThreads(BugWatchUpdater):
+    ...     def _getExternalBugTrackersAndWatches(
+    ...         self, bug_trackers, bug_watches):
+    ...         return [(ExternalBugTrackerForThreads(), bug_watches)]
+
+    >>> threaded_bug_watch_updater = BugWatchUpdaterForThreads(
+    ...     transaction, FakeLogger(output_file))
+    >>> threaded_bug_watch_updater.updateBugTrackers(
+    ...     batch_size=5, num_threads=10)
+
+    >>> for output in sorted(output_file.output.itervalues()):
+    ...     print "".join(output),
+    ...     print '--'
+    DEBUG No watches to update on http://bugs.example.com
+    --
+    DEBUG No watches to update on http://bugzilla.gnome.org/
+    --
+    DEBUG No watches to update on http://savannah.gnu.org/
+    --
+    DEBUG No watches to update on http://sourceforge.net/
+    --
+    DEBUG No watches to update on mailto:[e-mail]
+    --
+    DEBUG Using a global batch size of 5
+    DEBUG Skipping updating Ubuntu Bugzilla watches.
+    --
+    INFO Updating 2 watches for 2 bugs on http://example.com
+    getRemoteStatus(bug_id=u'304070')
+    getRemoteStatus(bug_id=u'3224')
+    --
+    INFO Updating 4 watches for 3 bugs on http://example.com
+    getRemoteStatus(bug_id=u'123543')
+    getRemoteStatus(bug_id=u'2000')
+    getRemoteStatus(bug_id=u'42')
+    --
+    INFO Updating 5 watches for 5 bugs on http://example.com
+    getRemoteStatus(bug_id=u'1')
+    getRemoteStatus(bug_id=u'101')
+    getRemoteStatus(bug_id=u'5')
+    getRemoteStatus(bug_id=u'6')
+    getRemoteStatus(bug_id=u'7')
+    --
+    INFO Updating 5 watches for 5 bugs on http://example.com
+    getRemoteStatus(bug_id=u'280883')
+    getRemoteStatus(bug_id=u'304014')
+    getRemoteStatus(bug_id=u'308994')
+    getRemoteStatus(bug_id=u'327452')
+    getRemoteStatus(bug_id=u'327549')
+    --

=== modified file 'lib/lp/bugs/scripts/checkwatches.py'
--- lib/lp/bugs/scripts/checkwatches.py	2009-11-19 12:49:04 +0000
+++ lib/lp/bugs/scripts/checkwatches.py	2009-11-27 16:20:37 +0000
@@ -6,8 +6,10 @@
 
 from copy import copy
 from datetime import datetime, timedelta
+import Queue as queue
 import socket
 import sys
+import threading
 import time
 
 import pytz
@@ -31,7 +33,7 @@
     ErrorReportingUtility, ScriptRequest)
 from canonical.launchpad.webapp.interfaces import IPlacelessAuthUtility
 from canonical.launchpad.webapp.interaction import (
-    setupInteraction, endInteraction)
+    setupInteraction, endInteraction, queryInteraction)
 from canonical.launchpad.webapp.publisher import canonical_url
 
 from lp.bugs import externalbugtracker
@@ -161,7 +163,21 @@
 
     ACCEPTABLE_TIME_SKEW = timedelta(minutes=10)
 
+    LOGIN = '[e-mail]'
+
     def __init__(self, txn, log=default_log, syncable_gnome_products=None):
+        """Initialize a BugWatchUpdater.
+
+        :param txn: A transaction manager on which `begin()`,
+            `abort()` and `commit()` can be called. Additionally, it
+            should be safe for different threads to use its methods to
+            manage their own transactions (i.e. with thread-local
+            storage).
+
+        :param log: An instance of `logging.Logger`, or something that
+            provides a similar interface.
+
+        """
         self.txn = txn
         self.log = log
 
@@ -171,96 +187,176 @@
         else:
             self._syncable_gnome_products = list(SYNCABLE_GNOME_PRODUCTS)
 
+        self._principal = (
+            getUtility(IPlacelessAuthUtility).getPrincipalByLogin(
+                self.LOGIN, want_password=False))
+
     def _login(self):
         """Set up an interaction as the Bug Watch Updater"""
-        auth_utility = getUtility(IPlacelessAuthUtility)
-        setupInteraction(
-            auth_utility.getPrincipalByLogin(
-                '[e-mail]', want_password=False),
-            login='[e-mail]')
+        setupInteraction(self._principal, login=self.LOGIN)
 
     def _logout(self):
         """Tear down the Bug Watch Updater Interaction."""
         endInteraction()
 
-    def updateBugTrackers(self, bug_tracker_names=None, batch_size=None):
-        """Update all the bug trackers that have watches pending.
-
-        If bug tracker names are specified in bug_tracker_names only
-        those bug trackers will be checked.
+    def _interactionDecorator(self, func):
+        """Wrap a function to ensure that it runs within an interaction.
+
+        If an interaction is already set up, this simply calls the
+        function. If no interaction exists, it will set one up, call the
+        function, then end the interaction.
+
+        This is intended to make sure the right thing happens whether or not
+        the function is run in a different thread.
         """
-        self.txn.begin()
+        def wrapper(*args, **kwargs):
+            if queryInteraction() is None:
+                self._login()
+                try:
+                    return func(*args, **kwargs)
+                finally:
+                    self._logout()
+            else:
+                return func(*args, **kwargs)
+        return wrapper
+
+    def _bugTrackerUpdaters(self, bug_tracker_names=None):
+        """Yields functions that can be used to update each bug tracker."""
+        # Set up an interaction as the Bug Watch Updater since the
+        # notification code expects a logged in user.
+        self._login()
+
         ubuntu_bugzilla = getUtility(ILaunchpadCelebrities).ubuntu_bugzilla
         # Save the name, so we can use it in other transactions.
         ubuntu_bugzilla_name = ubuntu_bugzilla.name
 
-        # Set up an interaction as the Bug Watch Updater since the
-        # notification code expects a logged in user.
-        self._login()
-
-        self.log.debug("Using a global batch size of %s" % batch_size)
-
         if bug_tracker_names is None:
             bug_tracker_names = [
                 bugtracker.name for bugtracker in getUtility(IBugTrackerSet)]
-        self.txn.commit()
+
+        def make_updater(bug_tracker_id):
+            """Returns a function that can update the given bug tracker."""
+            def updater(batch_size=None):
+                run = self._interactionDecorator(self.updateBugTracker)
+                return run(bug_tracker_id, batch_size)
+            return updater
+
         for bug_tracker_name in bug_tracker_names:
-            self.txn.begin()
-            bug_tracker = getUtility(IBugTrackerSet).getByName(
-                bug_tracker_name)
-
-            if not bug_tracker.active:
+            if bug_tracker_name == ubuntu_bugzilla_name:
+                # XXX: 2007-09-11 Graham Binns
+                #      We automatically ignore the Ubuntu Bugzilla
+                #      here as all its bugs have been imported into
+                #      Launchpad. Ideally we would have some means
+                #      to identify all bug trackers like this so
+                #      that hard-coding like this can be genericised
+                #      (Bug 138949).
                 self.log.debug(
-                    "Updates are disabled for bug tracker at %s" %
-                    bug_tracker.baseurl)
-                self.txn.abort()
-                continue
-
-            # Save the url for later, since we might need it to report an
-            # error after a transaction has been aborted.
-            bug_tracker_url = bug_tracker.baseurl
-            try:
-                if bug_tracker_name == ubuntu_bugzilla_name:
-                    # XXX: 2007-09-11 Graham Binns
-                    #      We automatically ignore the Ubuntu Bugzilla
-                    #      here as all its bugs have been imported into
-                    #      Launchpad. Ideally we would have some means
-                    #      to identify all bug trackers like this so
-                    #      that hard-coding like this can be genericised
-                    #      (Bug 138949).
+                    "Skipping updating Ubuntu Bugzilla watches.")
+            else:
+                bug_tracker = getUtility(IBugTrackerSet).getByName(
+                    bug_tracker_name)
+                if bug_tracker.active:
+                    yield make_updater(bug_tracker.id)
+                else:
                     self.log.debug(
-                        "Skipping updating Ubuntu Bugzilla watches.")
-                else:
-                    self.updateBugTracker(bug_tracker, batch_size)
+                        "Updates are disabled for bug tracker at %s" %
+                        bug_tracker.baseurl)
 
-                self.txn.commit()
-            except (KeyboardInterrupt, SystemExit):
-                # We should never catch KeyboardInterrupt or SystemExit.
-                raise
-            except Exception, error:
-                # If something unexpected goes wrong, we log it and
-                # continue: a failure shouldn't break the updating of
-                # the other bug trackers.
-                info = sys.exc_info()
-                properties = [
-                    ('bugtracker', bug_tracker_name),
-                    ('baseurl', bug_tracker_url)]
-                if isinstance(error, BugWatchUpdateError):
-                    self.error(
-                        str(error), properties=properties, info=info)
-                elif isinstance(error, socket.timeout):
-                    self.error(
-                        "Connection timed out when updating %s" %
-                        bug_tracker_url,
-                        properties=properties, info=info)
-                else:
-                    self.error(
-                        "An exception was raised when updating %s" %
-                        bug_tracker_url,
-                        properties=properties, info=info)
-                self.txn.abort()
         self._logout()
 
+    def updateBugTrackers(
+        self, bug_tracker_names=None, batch_size=None, num_threads=1):
+        """Update all the bug trackers that have watches pending.
+
+        If bug tracker names are specified in bug_tracker_names only
+        those bug trackers will be checked.
+
+        The updates are run in threads, so that long running updates
+        don't block progress. However, by default the number of
+        threads is 1, to help with testing.
+        """
+        self.log.debug("Using a global batch size of %s" % batch_size)
+
+        # Put all the work on the queue. This is simpler than drip-feeding the
+        # queue, and avoids a situation where a worker thread exits because
+        # there's no work left and the feeding thread hasn't been scheduled to
+        # add work to the queue.
+        work = queue.Queue()
+        for updater in self._bugTrackerUpdaters(bug_tracker_names):
+            work.put(updater)
+
+        # This will be run once in each worker thread.
+        def do_work():
+            while True:
+                try:
+                    job = work.get(block=False)
+                except queue.Empty:
+                    break
+                else:
+                    job(batch_size)
+
+        # Start and join the worker threads.
+        threads = []
+        for run in xrange(num_threads):
+            thread = threading.Thread(target=do_work)
+            thread.start()
+            threads.append(thread)
+        for thread in threads:
+            thread.join()
+
+    def updateBugTracker(self, bug_tracker, batch_size):
+        """Updates the given bug trackers's bug watches.
+
+        If there is an error, logs are updated, and the transaction is
+        aborted.
+
+        :param bug_tracker: An IBugTracker or the ID of one, so that this
+            method can be called from a different interaction.
+
+        :return: A boolean indicating if the operation was successful.
+        """
+        # Get the bug tracker.
+        if isinstance(bug_tracker, (int, long)):
+            bug_tracker = getUtility(IBugTrackerSet).get(bug_tracker)
+
+        # Save the name and url for later, since we might need it to report an
+        # error after a transaction has been aborted.
+        bug_tracker_name = bug_tracker.name
+        bug_tracker_url = bug_tracker.baseurl
+
+        try:
+            self.txn.begin()
+            self._updateBugTracker(bug_tracker, batch_size)
+            self.txn.commit()
+        except (KeyboardInterrupt, SystemExit):
+            # We should never catch KeyboardInterrupt or SystemExit.
+            raise
+        except Exception, error:
+            # If something unexpected goes wrong, we log it and
+            # continue: a failure shouldn't break the updating of
+            # the other bug trackers.
+            info = sys.exc_info()
+            properties = [
+                ('bugtracker', bug_tracker_name),
+                ('baseurl', bug_tracker_url)]
+            if isinstance(error, BugWatchUpdateError):
+                self.error(
+                    str(error), properties=properties, info=info)
+            elif isinstance(error, socket.timeout):
+                self.error(
+                    "Connection timed out when updating %s" %
+                    bug_tracker_url,
+                    properties=properties, info=info)
+            else:
+                self.error(
+                    "An exception was raised when updating %s" %
+                    bug_tracker_url,
+                    properties=properties, info=info)
+            self.txn.abort()
+            return False
+        else:
+            return True
+
     def forceUpdateAll(self, bug_tracker_name, batch_size):
         """Update all the watches for `bug_tracker_name`.
 
@@ -290,50 +386,16 @@
         bug_tracker.resetWatches()
         self.txn.commit()
 
-        # Take a copy of the bug tracker URL. If the transaction fails
-        # later we can't refer to the baseurl attribute of the bug
-        # tracker.
-        bug_tracker_url = bug_tracker.baseurl
-
         # Loop over the bug watches in batches as specificed by
         # batch_size until there are none left to update.
         self.log.info(
             "Updating %s watches on bug tracker '%s'" %
             (bug_tracker.watches.count(), bug_tracker_name))
-        iteration = 0
         has_watches_to_update = True
         while has_watches_to_update:
             self.txn.begin()
-            try:
-                self.updateBugTracker(bug_tracker, batch_size)
-                self.txn.commit()
-            except (KeyboardInterrupt, SystemExit):
-                # We should never catch KeyboardInterrupt or SystemExit.
-                raise
-            except Exception, error:
-                # If something unexpected goes wrong, we log it and
-                # continue: a failure shouldn't break the updating of
-                # the other bug trackers.
-                info = sys.exc_info()
-                properties = [
-                    ('bugtracker', bug_tracker_name),
-                    ('baseurl', bug_tracker_url)]
-                if isinstance(error, BugWatchUpdateError):
-                    self.error(
-                        str(error), properties=properties, info=info)
-                elif isinstance(error, socket.timeout):
-                    self.error(
-                        "Connection timed out when updating %s" %
-                        bug_tracker_url,
-                        properties=properties, info=info)
-                else:
-                    self.error(
-                        "An exception was raised when updating %s" %
-                        bug_tracker_url,
-                        properties=properties, info=info)
-                self.txn.abort()
+            if not self.updateBugTracker(bug_tracker, batch_size):
                 break
-
             watches_left = bug_tracker.getBugWatchesNeedingUpdate(23).count()
             self.log.info(
                 "%s watches left to check on bug tracker '%s'" %
@@ -409,7 +471,7 @@
 
         return trackers_and_watches
 
-    def updateBugTracker(self, bug_tracker, batch_size=None):
+    def _updateBugTracker(self, bug_tracker, batch_size=None):
         """Updates the given bug trackers's bug watches."""
         # XXX 2007-01-18 gmb:
         #     Once we start running checkwatches more frequently we need
@@ -1085,7 +1147,7 @@
                 "one bugtracker using this option will check all the "
                 "bugtrackers specified.")
         self.parser.add_option(
-            '-b', '--batch-size', action='store', dest='batch_size',
+            '-b', '--batch-size', action='store', type=int, dest='batch_size',
             help="Set the number of watches to be checked per bug "
                  "tracker in this run. If BATCH_SIZE is 0, all watches "
                  "on the bug tracker that are eligible for checking will "
@@ -1094,26 +1156,27 @@
             '--reset', action='store_true', dest='update_all',
             help="Update all the watches on the bug tracker, regardless of "
                  "whether or not they need checking.")
+        self.parser.add_option(
+            '--jobs', action='store', type=int, dest='jobs', default=1,
+            help=("The number of simulataneous jobs to run, %default by "
+                  "default."))
 
     def main(self):
         start_time = time.time()
 
         updater = BugWatchUpdater(self.txn, self.logger)
 
-        # Make sure batch_size is an integer or None.
-        batch_size = self.options.batch_size
-        if batch_size is not None:
-            batch_size = int(batch_size)
-
         if self.options.update_all and len(self.options.bug_trackers) > 0:
             # The user has requested that we update *all* the watches
             # for these bugtrackers
             for bug_tracker in self.options.bug_trackers:
-                updater.forceUpdateAll(bug_tracker, batch_size)
+                updater.forceUpdateAll(bug_tracker, self.options.batch_size)
         else:
             # Otherwise we just update those watches that need updating,
             # and we let the BugWatchUpdater decide which those are.
-            updater.updateBugTrackers(self.options.bug_trackers, batch_size)
+            updater.updateBugTrackers(
+                self.options.bug_trackers, self.options.batch_size,
+                self.options.jobs)
 
         run_time = time.time() - start_time
         self.logger.info("Time for this run: %.3f seconds." % run_time)

=== modified file 'lib/lp/bugs/scripts/tests/test_bugimport.py'
--- lib/lp/bugs/scripts/tests/test_bugimport.py	2009-10-21 18:46:29 +0000
+++ lib/lp/bugs/scripts/tests/test_bugimport.py	2009-11-27 16:20:37 +0000
@@ -883,10 +883,10 @@
 class TestBugWatchUpdater(BugWatchUpdater):
     """A mock `BugWatchUpdater` object."""
 
-    def updateBugTracker(self, bug_tracker):
+    def _updateBugTracker(self, bug_tracker):
         # Save the current bug tracker, so _getBugWatch can reference it.
         self.bugtracker = bug_tracker
-        super(TestBugWatchUpdater, self).updateBugTracker(bug_tracker)
+        super(TestBugWatchUpdater, self)._updateBugTracker(bug_tracker)
 
     def _getExternalBugTrackersAndWatches(self, bug_tracker, bug_watches):
         """See `BugWatchUpdater`."""
@@ -928,7 +928,7 @@
         # trigger a DB error, the second updates successfully.
         bug_tracker = TestBugTracker(test_bug_one, test_bug_two)
         bug_watch_updater = TestBugWatchUpdater(self.layer.txn)
-        bug_watch_updater.updateBugTracker(bug_tracker)
+        bug_watch_updater._updateBugTracker(bug_tracker)
         # We verify that the first bug watch didn't update the status,
         # and the second did.
         for bugtask in test_bug_one.bugtasks: