Hi Jeroen,

This was not a trivial one but your cover letter did a very good job at
explaining what you did.  Thanks for taking the time to write it.

I have a few comments/suggestions below, but r=me

 review approve
 
On Fri, 2010-03-05 at 17:37 +0000, Jeroen T. Vermeulen wrote:
> Jeroen T. Vermeulen has proposed merging lp:~jtv/launchpad/bug-286237 into lp:launchpad/devel.


> === modified file 'lib/lp/translations/browser/tests/test_baseexportview.py'
> --- lib/lp/translations/browser/tests/test_baseexportview.py	2010-02-19 17:06:01 +0000
> +++ lib/lp/translations/browser/tests/test_baseexportview.py	2010-03-05 17:38:34 +0000
> @@ -7,6 +7,7 @@
>  import transaction
>  import unittest
>  
> +from canonical.launchpad.interfaces.lpstorm import IMasterStore
>  from canonical.launchpad.webapp.servers import LaunchpadTestRequest
>  from canonical.testing import ZopelessDatabaseLayer
>  from lp.translations.browser.sourcepackage import (
> @@ -15,13 +16,14 @@
>      ProductSeriesTranslationsExportView)
>  from lp.translations.interfaces.translationfileformat import (
>      TranslationFileFormat)
> +from lp.translations.model.poexportrequest import POExportRequest
>  from lp.testing import TestCaseWithFactory
>  
> 
>  def wipe_queue(queue):
>      """Erase all export queue entries."""
> -    while queue.entry_count > 0:
> -        queue.popRequest()
> +    queue_ids = IMasterStore(POExportRequest).execute(

You don't need to store the queue ids here, do you?

> +        "DELETE FROM POExportRequest")
>  
> 
>  class BaseExportViewMixin(TestCaseWithFactory):
> 
> === modified file 'lib/lp/translations/doc/poexport-queue.txt'
> --- lib/lp/translations/doc/poexport-queue.txt	2010-02-19 16:02:16 +0000
> +++ lib/lp/translations/doc/poexport-queue.txt	2010-03-05 17:38:34 +0000
> @@ -275,7 +275,8 @@
>  
>  Once the queue is processed, the queue is empty again.
>  
> -    >>> process_queue(fake_transaction, logging.getLogger())
> +    >>> transaction.commit()
> +    >>> process_queue(transaction, logging.getLogger())
>      INFO:...Stored file at http://.../po_evolution-2.2.pot
>  
>      >>> export_request_set.entry_count
> @@ -333,10 +334,10 @@
>  
>      >>> export_request_set.addRequest(
>      ...     carlos, pofiles=[pofile], format=TranslationFileFormat.PO)
> -    >>> process_queue(fake_transaction, logging.getLogger())
> +    >>> transaction.commit()
> +    >>> process_queue(transaction, logging.getLogger())
>      INFO:root:Stored file at http://...eo.po
>  
> -    >>> transaction.commit()
>      >>> print get_newest_librarian_file().read()
>      # Esperanto translation for ...
>      ...
> @@ -354,10 +355,10 @@
>  
>      >>> export_request_set.addRequest(
>      ...     carlos, pofiles=[pofile], format=TranslationFileFormat.POCHANGED)
> -    >>> process_queue(fake_transaction, logging.getLogger())
> +    >>> transaction.commit()
> +    >>> process_queue(transaction, logging.getLogger())
>      INFO:root:Stored file at http://...eo.po
>  
> -    >>> transaction.commit()
>      >>> print get_newest_librarian_file().read()
>      # IMPORTANT: This file does NOT contain a complete PO file structure.
>      # DO NOT attempt to import this file back into Launchpad.
> 
> === modified file 'lib/lp/translations/doc/poexport-request-productseries.txt'
> --- lib/lp/translations/doc/poexport-request-productseries.txt	2010-02-19 16:54:42 +0000
> +++ lib/lp/translations/doc/poexport-request-productseries.txt	2010-03-05 17:38:34 +0000
> @@ -35,11 +35,12 @@
>  Now we request that the queue be processed.
>  
>      >>> import logging
> -    >>> from lp.testing.faketransaction import FakeTransaction
> +    >>> import transaction
>      >>> from lp.translations.scripts.po_export_queue import process_queue
>      >>> logger = MockLogger()
>      >>> logger.setLevel(logging.DEBUG)
> -    >>> process_queue(FakeTransaction(), logger)
> +    >>> transaction.commit()
> +    >>> process_queue(transaction, logger)
>      log> Exporting objects for ..., related to template evolution-2.2 in
>      Evolution trunk
>      log> Exporting objects for ..., related to template evolution-2.2-test in
> 
> === modified file 'lib/lp/translations/doc/poexport-request.txt'
> --- lib/lp/translations/doc/poexport-request.txt	2010-02-19 16:54:42 +0000
> +++ lib/lp/translations/doc/poexport-request.txt	2010-03-05 17:38:34 +0000
> @@ -53,9 +53,10 @@
>  
>  Now we request that the queue be processed.
>  
> -    >>> from lp.testing.faketransaction import FakeTransaction
> +    >>> import transaction
>      >>> from lp.translations.scripts.po_export_queue import process_queue
> -    >>> process_queue(FakeTransaction(), MockLogger())
> +    >>> transaction.commit()
> +    >>> process_queue(transaction, MockLogger())
>      log> Exporting objects for Happy Downloader, related to template pmount
>      in Ubuntu Hoary package "pmount"
>      log> Stored file at http://.../launchpad-export.tar.gz
> @@ -185,7 +186,8 @@
>      >>> from lp.translations.interfaces.translationfileformat import (
>      ...     TranslationFileFormat)
>      >>> request_set.addRequest(person, None, [cs], TranslationFileFormat.MO)
> -    >>> process_queue(FakeTransaction(), MockLogger())
> +    >>> transaction.commit()

Do you think it'd be worth adding comments explaining why we need all
these commits?

> +    >>> process_queue(transaction, MockLogger())
>      log> Exporting objects for Happy Downloader, related to template pmount
>      in Ubuntu Hoary package "pmount"
>      log> Stored file at http://.../cs_LC_MESSAGES_pmount.mo
> 
> === added file 'lib/lp/translations/doc/poexportqueue-replication-lag.txt'
> --- lib/lp/translations/doc/poexportqueue-replication-lag.txt	1970-01-01 00:00:00 +0000
> +++ lib/lp/translations/doc/poexportqueue-replication-lag.txt	2010-03-05 17:38:34 +0000
> @@ -0,0 +1,89 @@
> += Replication Lag and the Export Queue =
> +
> +Due to replication lag it's possible for the export queue to see a
> +request on the slave store that it actually just removed from the master
> +store.
> +
> +We start our story with an empty export queue.
> +
> +    >>> from datetime import timedelta
> +    >>> import transaction
> +    >>> from zope.component import getUtility
> +    >>> from canonical.launchpad.interfaces.lpstorm import IMasterStore
> +    >>> from lp.translations.interfaces.poexportrequest import (
> +    ...     IPOExportRequestSet)
> +    >>> from lp.translations.interfaces.pofile import IPOFile
> +    >>> from lp.translations.model.poexportrequest import POExportRequest
> +    >>> query = IMasterStore(POExportRequest).execute(
> +    ...     "DELETE FROM POExportRequest")
> +
> +    >>> queue = getUtility(IPOExportRequestSet)
> +
> +We have somebody making an export request.
> +
> +    >>> requester = factory.makePersonNoCommit(
> +    ...     <email address hidden>', name='punter')
> +
> +    >>> template1 = factory.makePOTemplate()
> +    >>> pofile1_be = factory.makePOFile('be', potemplate=template1)
> +    >>> pofile1_ja = factory.makePOFile('ja', potemplate=template1)
> +    >>> queue.addRequest(requester, template1, [pofile1_be, pofile1_ja])
> +    >>> query = IMasterStore(POExportRequest).execute(
> +    ...     "UPDATE POExportRequest SET date_created = '2010-01-10'::date")
> +
> +Later, a different and separate request follows.
> +
> +    >>> template2 = factory.makePOTemplate()
> +    >>> pofile2_se = factory.makePOFile('se', potemplate=template2)
> +    >>> pofile2_ga = factory.makePOFile('ga', potemplate=template2)
> +    >>> queue.addRequest(requester, template2, [pofile2_se, pofile2_ga])
> +
> +The database is replicated in this state.
> +
> +    >>> transaction.commit()
> +
> +getRequest at this point returns the oldest request.
> +
> +    >>> def summarize_request(request_tuple):
> +    ...     """Summarize files in export request."""
> +    ...     person, sources, format, request_ids = request_tuple
> +    ...     summary = []
> +    ...     for source in sources:
> +    ...         if IPOFile.providedBy(source):
> +    ...             summary.append(source.language.code)
> +    ...         else:
> +    ...             summary.append('(template)')
> +    ...     for entry in sorted(summary):
> +    ...         print entry
> +
> +    >>> summarize_request(queue.getRequest())
> +    (template)
> +    be
> +    ja
> +
> +It doesn't modify the queue, so it'd say the same thing again if we
> +were to ask again.
> +
> +    >>> repeated_request = queue.getRequest()
> +    >>> summarize_request(repeated_request)
> +    (template)
> +    be
> +    ja
> +
> +The first request is removed from the master store after processing, but
> +not yet from the slave store.  (Since this test is all one session, we
> +can reproduce this by not committing the removal).  The second request
> +is still technically on the queue, but no longer "live."
> +
> +    >>> person, sources, format, request_ids = repeated_request
> +    >>> print len(request_ids)
> +    3
> +    >>> queue.removeRequest(request_ids)
> +
> +In this state, despite the replication lag, getRequest is smart enough
> +to return the second request, not the first.
> +
> +    >>> summarize_request(queue.getRequest())
> +    (template)
> +    ga
> +    se

This is one case where I think a unittest would be a better alternative;
we'd move the fixture setup into the setUp() method, create a few helper
methods, custom assertions and in the end the actual test would be no
more than a few lines of code, and very easy to spot. As a doctest I
have to go through all the fixture setup and helper methods just to find
where the actual test is.

I won't ask you to turn it into a unittest, although you're welcome to
do so if you want. ;)

> === modified file 'lib/lp/translations/model/poexportrequest.py'
> --- lib/lp/translations/model/poexportrequest.py	2010-02-19 17:06:01 +0000
> +++ lib/lp/translations/model/poexportrequest.py	2010-03-05 17:38:34 +0000
> @@ -112,36 +115,74 @@
>                      existing.id IS NULL
>                  """ % query_params)
>  
> -    def popRequest(self):
> -        """See `IPOExportRequestSet`."""
> -        try:
> -            request = POExportRequest.select(limit=1, orderBy='id')[0]
> -        except IndexError:
> -            return None, None, None
> -
> -        person = request.person
> -        format = request.format
> -
> -        query = """
> -            person = %s AND
> -            format = %s AND
> -            date_created = (
> -                SELECT date_created
> -                FROM POExportRequest
> -                ORDER BY id
> -                LIMIT 1)""" % sqlvalues(person, format)
> -        requests = POExportRequest.select(query, orderBy='potemplate')
> -        objects = []
> -
> -        for request in requests:
> -            if request.pofile is not None:
> -                objects.append(request.pofile)
> -            else:
> -                objects.append(request.potemplate)
> -
> -            POExportRequest.delete(request.id)
> -
> -        return person, objects, format
> +    def _getOldestLiveRequest(self):
> +        """Return the oldest live request on the master store.
> +
> +        Due to replication lag, the master store is always a little
> +        ahead of the slave store that exports come from.
> +        """
> +        master_store = IMasterStore(POExportRequest)
> +        sorted_by_id = master_store.find(POExportRequest).order_by(
> +            POExportRequest.id)
> +        return sorted_by_id.first()
> +
> +    def _getHeadRequest(self):
> +        """Return oldest request on the queue."""
> +        # Due to replication lag, it's possible that the slave store
> +        # still has copies of requests that have already been completed
> +        # and deleted from the master store.  So first get the oldest
> +        # request that is "live," i.e. still present on the master
> +        # store.
> +        oldest_live = self._getOldestLiveRequest()
> +        if oldest_live is None:
> +            return None
> +
> +        # Look only for requests on the slave store that are certainly
> +        # live.
> +        slave_store = ISlaveStore(POExportRequest)
> +        live_requests = slave_store.find(
> +            POExportRequest,
> +            POExportRequest.id >= oldest_live.id)

Is this not always going to return the same object (albeit coming from a
different store) that was returned by self._getOldestLiveRequest()?  In
that case it'd be much simpler to just return
ISlaveStore(POExportRequest).get(oldest_live.id).

> +
> +        return live_requests.order_by(POExportRequest.id).first()
> +
> +    def getRequest(self):
> +        """See `IPOExportRequestSet`."""
> +        # Exports happen off the slave store.  To ensure that export
> +        # does not happen until requests have been replicated to the
> +        # slave, they are read primarily from the slave even though they
> +        # are deleted on the master afterwards.
> +        head = self._getHeadRequest()
> +        if head is None:
> +            return None, None, None, None
> +
> +        requests = ISlaveStore(POExportRequest).find(
> +            POExportRequest,
> +            POExportRequest.person == head.person,
> +            POExportRequest.format == head.format,
> +            POExportRequest.date_created == head.date_created).order_by(
> +                POExportRequest.potemplateID)
> +
> +        summary = [
> +            (request.id, request.pofile or request.potemplate)
> +            for request in requests
> +            ]
> +
> +        sources = [source for request_id, source in summary]
> +        request_ids = [request_id for request_id, source in summary]
> +
> +        return head.person, sources, head.format, request_ids
> +
> +    def removeRequest(self, request_ids):
> +        """See `IPOExportRequestSet`."""
> +        if len(request_ids) > 0:
> +            # Storm 0.15 does not have direct support for deleting based
> +            # on is_in expressions and such, so do it the hard way.
> +            ids_string = ', '.join(sqlvalues(*request_ids))
> +            IMasterStore(POExportRequest).execute("""
> +                DELETE FROM POExportRequest
> +                WHERE id in (%s)
> +                """ % ids_string)
>  
> 
>  class POExportRequest(SQLBase):
> @@ -152,6 +193,7 @@
>      person = ForeignKey(
>          dbName='person', foreignKey='Person',
>          storm_validator=validate_public_person, notNull=True)
> +    date_created = UtcDateTimeCol(dbName='date_created', default=DEFAULT)
>      potemplate = ForeignKey(dbName='potemplate', foreignKey='POTemplate',
>          notNull=True)
>      pofile = ForeignKey(dbName='pofile', foreignKey='POFile')
> 
> === modified file 'lib/lp/translations/scripts/po_export_queue.py'
> --- lib/lp/translations/scripts/po_export_queue.py	2009-08-17 13:42:00 +0000
> +++ lib/lp/translations/scripts/po_export_queue.py	2010-03-05 17:38:34 +0000
> @@ -373,32 +373,17 @@
>  
> 
>  def process_queue(transaction_manager, logger):
> -    """Process each request in the translation export queue.
> +    """Process all requests in the translation export queue.
>  
> -    Each item is removed from the queue as it is processed, we only handle
> -    one request with each function call.
> +    Each item is removed from the queue as it is processed.
>      """
>      request_set = getUtility(IPOExportRequestSet)
>  
> -    request = request_set.popRequest()
> -
> -    if None in request:
> -        # Any value is None and we must have all values as not None to have
> -        # something to process...
> -        return
> -
> -    person, objects, format = request
> -
> -    try:
> +    request = request_set.getRequest()
> +    while None not in request:

I'd rather see this as

    while request != (None, None, None, None):
        ...

instead.  What do you think?

Either that or a comment, like the one we had before.

> +        person, objects, format, request_ids = request
>          process_request(person, objects, format, logger)
> -    except psycopg2.Error:
> -        # We had a DB error, we don't try to recover it here, just exit
> -        # from the script and next run will retry the export.
> -        logger.error(
> -            "A DB exception was raised when exporting files for %s" % (
> -                person.displayname),
> -            exc_info=True)
> -        transaction_manager.abort()
> -    else:
> -        # Apply all changes.
> +        request_set.removeRequest(request_ids)
>          transaction_manager.commit()
> +
> +        request = request_set.getRequest()

-- 
Guilherme Salgado <email address hidden>