Merge lp:~wgrant/launchpad/renovate-deathrow into lp:launchpad

[William Grant]

Unmerged revisions

16519. By William Grant on 2013-03-07

Rework SourceDeathRow to use a more efficient query.

16518. By William Grant on 2013-03-07

Implement a hopefully correct new deathrow algorithm, and fix the tests to basically cope.

16517. By William Grant on 2013-03-06

Split out and start testing reapable file calculation.

16516. By William Grant on 2013-03-01

Use component/SPN IDs where possible.

16515. By William Grant on 2013-03-01

Start rewriting process-death-row as a LoopTuner. Currently it just removes any file referenced by a removable publication, without any of the useful checks.

=== modified file 'lib/lp/archivepublisher/deathrow.py'
--- lib/lp/archivepublisher/deathrow.py	2012-08-17 11:15:35 +0000
+++ lib/lp/archivepublisher/deathrow.py	2016-09-24 06:41:02 +0000
@@ -6,31 +6,65 @@
 """
 __metaclass__ = type
 
-import datetime
 import logging
-import os
 
-import pytz
+from storm.expr import (
+    And,
+    Exists,
+    Not,
+    Select,
+    SQL,
+    )
+from storm.locals import (
+    Int,
+    Unicode,
+    )
+import transaction
 
 from lp.archivepublisher.config import getPubConfig
 from lp.archivepublisher.diskpool import DiskPool
+from lp.registry.model.sourcepackagename import SourcePackageName
 from lp.services.database.constants import UTC_NOW
-from lp.services.database.sqlbase import sqlvalues
+from lp.services.database.lpstorm import IStore
+from lp.services.database.sqlbase import convert_storm_clause_to_string
+from lp.services.database.stormexpr import Values
+from lp.services.librarian.model import (
+    LibraryFileAlias,
+    LibraryFileContent,
+    )
+from lp.services.looptuner import TunableLoop
 from lp.soyuz.enums import ArchivePurpose
 from lp.soyuz.interfaces.publishing import (
-    IBinaryPackagePublishingHistory,
-    inactive_publishing_status,
-    ISourcePackagePublishingHistory,
     MissingSymlinkInPool,
     NotInPool,
     )
+from lp.soyuz.model.binarypackagebuild import BinaryPackageBuild
+from lp.soyuz.model.binarypackagerelease import BinaryPackageRelease
+from lp.soyuz.model.component import Component
+from lp.soyuz.model.files import (
+    BinaryPackageFile,
+    SourcePackageReleaseFile,
+    )
 from lp.soyuz.model.publishing import (
     BinaryPackagePublishingHistory,
     SourcePackagePublishingHistory,
     )
 
 
-def getDeathRow(archive, log, pool_root_override):
+# Storm representation of calculateReapableFiles' CTE. Not a real table.
+class Interesting:
+    __storm_table__ = 'interesting'
+    # Meaningless, but get_cls_info needs it.
+    __storm_primary__ = (
+        'component_id', 'sourcepackagename_id', 'filename', 'md5')
+
+    component_id = Int(name='component')
+    sourcepackagename_id = Int(name='sourcepackagename')
+    filename = Unicode()
+    md5 = Unicode()
+
+
+def getDeathRows(archive, log, pool_root_override, dryrun):
     """Return a Deathrow object for the archive supplied.
 
     :param archive: Use the publisher config for this archive to derive the
@@ -58,10 +92,12 @@
     dp = DiskPool(pool_root, pubconf.temproot, diskpool_log)
 
     log.debug("Preparing death row.")
-    return DeathRow(archive, dp, log)
-
-
-class DeathRow:
+    return (
+        SourceDeathRow(archive, dp, log, dryrun),
+        BinaryDeathRow(archive, dp, log, dryrun))
+
+
+class DeathRow(TunableLoop):
     """A Distribution Archive Removal Processor.
 
     DeathRow will remove archive files from disk if they are marked for
@@ -69,227 +105,180 @@
     by other packages.
     """
 
-    def __init__(self, archive, diskpool, logger):
+    maximum_chunk_size = 5000
+
+    def __init__(self, archive, diskpool, logger, dry_run=False):
+        super(DeathRow, self).__init__(logger)
         self.archive = archive
         self.diskpool = diskpool
         self._removeFile = diskpool.removeFile
         self.logger = logger
-
-    def reap(self, dry_run=False):
-        """Reap packages that should be removed from the distribution.
-
-        Looks through all packages that are in condemned states and
-        have scheduleddeletiondate is in the past, try to remove their
-        files from the archive pool (which may be impossible if they are
-        used by other packages which are published), and mark them as
-        removed."""
-        if dry_run:
-            # Don't actually remove the files if we are dry running
-            def _mockRemoveFile(cn, sn, fn):
-                self.logger.debug("(Not really!) removing %s %s/%s" %
-                                  (cn, sn, fn))
-                fullpath = self.diskpool.pathFor(cn, sn, fn)
-                if not os.path.exists(fullpath):
-                    raise NotInPool
-                return os.lstat(fullpath).st_size
-            self._removeFile = _mockRemoveFile
-
-        source_files, binary_files = self._collectCondemned()
-        records = self._tryRemovingFromDisk(source_files, binary_files)
-        self._markPublicationRemoved(records)
-
-    def _collectCondemned(self):
-        """Return the condemned source and binary publications as a tuple.
-
-        Return all the `SourcePackagePublishingHistory` and
-        `BinaryPackagePublishingHistory` records that are eligible for
-        removal ('condemned') where the source/binary package that they
-        refer to is not published somewhere else.
-
-        Both sources and binaries are lists.
-        """
-        sources = SourcePackagePublishingHistory.select("""
-            SourcePackagePublishingHistory.archive = %s AND
-            SourcePackagePublishingHistory.scheduleddeletiondate < %s AND
-            SourcePackagePublishingHistory.dateremoved IS NULL AND
-            NOT EXISTS (
-              SELECT 1 FROM sourcepackagepublishinghistory as spph
-              WHERE
-                  SourcePackagePublishingHistory.sourcepackagerelease =
-                      spph.sourcepackagerelease AND
-                  spph.archive = %s AND
-                  spph.status NOT IN %s)
-        """ % sqlvalues(self.archive, UTC_NOW, self.archive,
-                        inactive_publishing_status), orderBy="id")
-        self.logger.debug("%d Sources" % sources.count())
-
-        binaries = BinaryPackagePublishingHistory.select("""
-            BinaryPackagePublishingHistory.archive = %s AND
-            BinaryPackagePublishingHistory.scheduleddeletiondate < %s AND
-            BinaryPackagePublishingHistory.dateremoved IS NULL AND
-            NOT EXISTS (
-              SELECT 1 FROM binarypackagepublishinghistory as bpph
-              WHERE
-                  BinaryPackagePublishingHistory.binarypackagerelease =
-                      bpph.binarypackagerelease AND
-                  bpph.archive = %s AND
-                  bpph.status NOT IN %s)
-        """ % sqlvalues(self.archive, UTC_NOW, self.archive,
-                        inactive_publishing_status), orderBy="id")
-        self.logger.debug("%d Binaries" % binaries.count())
-
-        return (sources, binaries)
-
-    def canRemove(self, publication_class, filename, file_md5):
-        """Check if given (filename, MD5) can be removed from the pool.
-
-        Check the archive reference-counter implemented in:
-        `SourcePackagePublishingHistory` or
-        `BinaryPackagePublishingHistory`.
-
-        Only allow removal of unnecessary files.
-        """
-        clauses = []
-        clauseTables = []
-
-        if ISourcePackagePublishingHistory.implementedBy(
-            publication_class):
-            clauses.append("""
-                SourcePackagePublishingHistory.archive = %s AND
-                SourcePackagePublishingHistory.dateremoved is NULL AND
-                SourcePackagePublishingHistory.sourcepackagerelease =
-                    SourcePackageReleaseFile.sourcepackagerelease AND
-                SourcePackageReleaseFile.libraryfile = LibraryFileAlias.id
-            """ % sqlvalues(self.archive))
-            clauseTables.append('SourcePackageReleaseFile')
-        elif IBinaryPackagePublishingHistory.implementedBy(
-            publication_class):
-            clauses.append("""
-                BinaryPackagePublishingHistory.archive = %s AND
-                BinaryPackagePublishingHistory.dateremoved is NULL AND
-                BinaryPackagePublishingHistory.binarypackagerelease =
-                    BinaryPackageFile.binarypackagerelease AND
-                BinaryPackageFile.libraryfile = LibraryFileAlias.id
-            """ % sqlvalues(self.archive))
-            clauseTables.append('BinaryPackageFile')
+        self.dry_run = dry_run
+
+        self.bytes = 0
+        self.start_at = 1
+        self.store = IStore(SourcePackagePublishingHistory)
+
+    def isDone(self):
+        return self.findPubs().is_empty()
+
+    def __call__(self, chunk_size):
+        pubs = list(self.findPubs()[:chunk_size])
+        files = self.calculateReapableFiles(pubs)
+
+        # The remaining files aren't referenced by other publications,
+        # so remove them from disk.
+        for comp, spn, filename in files:
+            if not self.dry_run:
+                try:
+                    self.bytes += self.diskpool.removeFile(comp, spn, filename)
+                except NotInPool as info:
+                    # It's safe for us to let this slide because it means that
+                    # the file is already gone.
+                    self.logger.debug(str(info))
+                except MissingSymlinkInPool as info:
+                    # This one is a little more worrying, because an expected
+                    # symlink has vanished from the pool/ (could be a code
+                    # mistake) but there is nothing we can do about it at this
+                    # point.
+                    self.logger.warn(str(info))
+            else:
+                self.logger.debug("Not removing %s (dry run)", filename)
+        for pub in pubs:
+            pub.dateremoved = UTC_NOW
+        self.start_at = pubs[-1].id + 1
+
+        if not self.dry_run:
+            transaction.commit()
         else:
-            raise AssertionError("%r is not supported." % publication_class)
-
-        clauses.append("""
-           LibraryFileAlias.content = LibraryFileContent.id AND
-           LibraryFileAlias.filename = %s AND
-           LibraryFileContent.md5 = %s
-        """ % sqlvalues(filename, file_md5))
-        clauseTables.extend(
-            ['LibraryFileAlias', 'LibraryFileContent'])
-
-        all_publications = publication_class.select(
-            " AND ".join(clauses), clauseTables=clauseTables)
-
-        right_now = datetime.datetime.now(pytz.timezone('UTC'))
-        for pub in all_publications:
-            # Deny removal if any reference is still active.
-            if pub.status not in inactive_publishing_status:
-                return False
-            # Deny removal if any reference wasn't dominated yet.
-            if pub.scheduleddeletiondate is None:
-                return False
-            # Deny removal if any reference is still in 'quarantine'.
-            if pub.scheduleddeletiondate > right_now:
-                return False
-
-        return True
-
-    def _tryRemovingFromDisk(self, condemned_source_files,
-                             condemned_binary_files):
-        """Take the list of publishing records provided and unpublish them.
-
-        You should only pass in entries you want to be unpublished because
-        this will result in the files being removed if they're not otherwise
-        in use.
-        """
-        bytes = 0
-        condemned_files = set()
-        condemned_records = set()
-        considered_files = set()
-        details = {}
-
-        def checkPubRecord(pub_record, publication_class):
-            """Check if the publishing record can be removed.
-
-            It can only be removed if all files in its context are not
-            referred to any other 'published' publishing records.
-
-            See `canRemove` for more information.
-            """
-            for pub_file in pub_record.files:
-                filename = pub_file.libraryfilealiasfilename
-                file_md5 = pub_file.libraryfilealias.content.md5
-
-                self.logger.debug("Checking %s (%s)" % (filename, file_md5))
-
-                # Calculating the file path in pool.
-                pub_file_details = (
-                    pub_file.libraryfilealiasfilename,
-                    pub_file.sourcepackagename,
-                    pub_file.componentname,
-                    )
-                file_path = self.diskpool.pathFor(*pub_file_details)
-
-                # Check if the LibraryFileAlias in question was already
-                # verified. If the verification was already made and the
-                # file is condemned queue the publishing record for removal
-                # otherwise just continue the iteration.
-                if (filename, file_md5) in considered_files:
-                    self.logger.debug("Already verified.")
-                    if file_path in condemned_files:
-                        condemned_records.add(pub_file.publishing_record)
-                    continue
-                considered_files.add((filename, file_md5))
-
-                # Check if the removal is allowed, if not continue.
-                if not self.canRemove(publication_class, filename, file_md5):
-                    self.logger.debug("Cannot remove.")
-                    continue
-
-                # Update local containers, in preparation to file removal.
-                details.setdefault(file_path, pub_file_details)
-                condemned_files.add(file_path)
-                condemned_records.add(pub_file.publishing_record)
-
-        # Check source and binary publishing records.
-        for pub_record in condemned_source_files:
-            checkPubRecord(pub_record, SourcePackagePublishingHistory)
-        for pub_record in condemned_binary_files:
-            checkPubRecord(pub_record, BinaryPackagePublishingHistory)
-
-        self.logger.info(
-            "Removing %s files marked for reaping" % len(condemned_files))
-
-        for condemned_file in sorted(condemned_files, reverse=True):
-            file_name, source_name, component_name = details[condemned_file]
-            try:
-                bytes += self._removeFile(
-                    component_name, source_name, file_name)
-            except NotInPool as info:
-                # It's safe for us to let this slide because it means that
-                # the file is already gone.
-                self.logger.debug(str(info))
-            except MissingSymlinkInPool as info:
-                # This one is a little more worrying, because an expected
-                # symlink has vanished from the pool/ (could be a code
-                # mistake) but there is nothing we can do about it at this
-                # point.
-                self.logger.warn(str(info))
-
-        self.logger.info("Total bytes freed: %s" % bytes)
-
-        return condemned_records
-
-    def _markPublicationRemoved(self, condemned_records):
-        # Now that the os.remove() calls have been made, simply let every
-        # now out-of-date record be marked as removed.
-        self.logger.debug("Marking %s condemned packages as removed." %
-                          len(condemned_records))
-        for record in condemned_records:
-            record.dateremoved = UTC_NOW
+            transaction.abort()
+
+    def calculateReapableFiles(self, pubs):
+        files = {}
+
+        # Find all the files that we might remove because of these
+        # publications.
+        for pub in pubs:
+            spr = self.getSourcePackageRelease(pub)
+            for file in self.getFiles(pub):
+                key = (
+                    pub.componentID, spr.sourcepackagenameID,
+                    file.libraryfile.filename,
+                    file.libraryfile.content.md5)
+                if key not in files:
+                    files[key] = set()
+
+        interesting_cols = [
+            ('component', 'integer'),
+            ('sourcepackagename', 'integer'),
+            ('filename', 'text'),
+            ('md5', 'text'),
+            ]
+        interesting_values = Values(None, interesting_cols, files.keys())
+
+        # Exclude any files that are referenced by other unremoved
+        # publications. They need to stay on disk now and will be
+        # removed with their final publication later.
+        # We find unremoved publications that aren't in the current
+        # batch, with files with the same
+        # (archive, component, sourcepackagename, filename, md5).
+        has_other_pubs = self.store.with_(
+            SQL('interesting(component, sourcepackagename, filename, md5) AS '
+                + convert_storm_clause_to_string(interesting_values))
+            ).find(
+                (Interesting.component_id, Interesting.sourcepackagename_id,
+                 Interesting.filename, Interesting.md5),
+                self.getOtherPubFilter(pubs),
+            ).config(distinct=True)
+        for key in has_other_pubs:
+            del files[key]
+
+        return [
+            (Component.get(c).name, SourcePackageName.get(s).name, f)
+            for (c, s, f, m) in files]
+
+
+class SourceDeathRow(DeathRow):
+
+    def findPubs(self):
+        return self.store.find(
+            SourcePackagePublishingHistory,
+            SourcePackagePublishingHistory.id >= self.start_at,
+            SourcePackagePublishingHistory.archive == self.archive,
+            SourcePackagePublishingHistory.scheduleddeletiondate < UTC_NOW,
+            SourcePackagePublishingHistory.dateremoved == None)
+
+    def getOtherPubFilter(self, pubs):
+        spr_select = Select(
+            SourcePackageReleaseFile.sourcepackagereleaseID,
+            tables=[
+                LibraryFileAlias, LibraryFileContent,
+                SourcePackageReleaseFile],
+            where=And(
+                LibraryFileAlias.filename == Interesting.filename,
+                LibraryFileContent.md5 == Interesting.md5,
+                LibraryFileContent.id == LibraryFileAlias.contentID,
+                LibraryFileAlias.id == SourcePackageReleaseFile.libraryfileID))
+        return Exists(Select(
+            1, tables=[SourcePackagePublishingHistory],
+            where=And(
+                SourcePackagePublishingHistory.sourcepackagereleaseID.is_in(
+                    spr_select),
+                # Filter conditions
+                SourcePackagePublishingHistory.componentID ==
+                    Interesting.component_id,
+                SourcePackagePublishingHistory.sourcepackagenameID ==
+                    Interesting.sourcepackagename_id,
+                SourcePackagePublishingHistory.archive == self.archive,
+                SourcePackagePublishingHistory.dateremoved == None,
+                Not(
+                    SourcePackagePublishingHistory.id.is_in(
+                        [pub.id for pub in pubs])))))
+
+    def getSourcePackageRelease(self, pub):
+        return pub.sourcepackagerelease
+
+    def getFiles(self, pub):
+        return pub.sourcepackagerelease.files
+
+
+class BinaryDeathRow(DeathRow):
+
+    def findPubs(self):
+        return self.store.find(
+            BinaryPackagePublishingHistory,
+            BinaryPackagePublishingHistory.id >= self.start_at,
+            BinaryPackagePublishingHistory.archive == self.archive,
+            BinaryPackagePublishingHistory.scheduleddeletiondate < UTC_NOW,
+            BinaryPackagePublishingHistory.dateremoved == None)
+
+    def getOtherPubQuery(self, pubs):
+        origin = [
+            BinaryPackagePublishingHistory, BinaryPackageRelease,
+            BinaryPackageBuild, BinaryPackageFile]
+        filter = [
+            # Join conditions
+            BinaryPackageRelease.id ==
+                BinaryPackagePublishingHistory.binarypackagereleaseID,
+            BinaryPackageBuild.id == BinaryPackageRelease.buildID,
+            BinaryPackageFile.binarypackagereleaseID ==
+                BinaryPackageRelease.id,
+            LibraryFileAlias.id == BinaryPackageFile.libraryfileID,
+            # Filter conditions
+            BinaryPackagePublishingHistory.componentID ==
+                Interesting.component_id,
+            BinaryPackageBuild.source_package_name_id ==
+                Interesting.sourcepackagename_id,
+            BinaryPackagePublishingHistory.archive == self.archive,
+            BinaryPackagePublishingHistory.dateremoved == None,
+            Not(
+                BinaryPackagePublishingHistory.id.is_in(
+                    [pub.id for pub in pubs])),
+            ]
+        return origin, filter
+
+    def getSourcePackageRelease(self, pub):
+        return pub.binarypackagerelease.build.source_package_release
+
+    def getFiles(self, pub):
+        return pub.binarypackagerelease.files

=== modified file 'lib/lp/archivepublisher/scripts/processdeathrow.py'
--- lib/lp/archivepublisher/scripts/processdeathrow.py	2016-09-19 11:19:16 +0000
+++ lib/lp/archivepublisher/scripts/processdeathrow.py	2016-09-24 06:41:02 +0000
@@ -15,6 +15,7 @@
     ]
 
 
+<<<<<<< TREE
 from lp.archivepublisher.deathrow import getDeathRow
 from lp.archivepublisher.scripts.base import PublisherScript
 from lp.services.limitedlist import LimitedList
@@ -25,6 +26,16 @@
 
 
 class DeathRowProcessor(PublisherScript):
+=======
+from zope.component import getUtility
+
+from lp.archivepublisher.deathrow import getDeathRows
+from lp.registry.interfaces.distribution import IDistributionSet
+from lp.services.scripts.base import LaunchpadCronScript
+
+
+class DeathRowProcessor(LaunchpadCronScript):
+>>>>>>> MERGE-SOURCE
 
     def add_my_options(self):
         self.parser.add_option(
@@ -59,10 +70,11 @@
         the operation just executed, i.e, commits successful runs and aborts
         runs with errors. It also respects 'dry-run' command-line option.
         """
-        death_row = getDeathRow(
-            archive, self.logger, self.options.pool_root)
+        sdr, bdr = getDeathRows(
+            archive, self.logger, self.options.pool_root, self.options.dry_run)
         self.logger.debug(
             "Unpublishing death row for %s." % archive.displayname)
+<<<<<<< TREE
         set_request_started(
             request_statements=LimitedList(10000),
             txn=self.txn, enable_timeout=False)
@@ -81,3 +93,7 @@
                 self.txn.commit()
         finally:
             clear_request_started()
+=======
+        sdr.run()
+        bdr.run()
+>>>>>>> MERGE-SOURCE

=== modified file 'lib/lp/archivepublisher/tests/deathrow.txt'
--- lib/lp/archivepublisher/tests/deathrow.txt	2011-04-05 08:28:30 +0000
+++ lib/lp/archivepublisher/tests/deathrow.txt	2016-09-24 06:41:02 +0000
@@ -16,12 +16,20 @@
 The no-operation use case, reflects the sampledata status.
 
     >>> from lp.services.log.logger import FakeLogger
-    >>> from lp.archivepublisher.deathrow import DeathRow
+    >>> from lp.archivepublisher.deathrow import (
+    ...     BinaryDeathRow,
+    ...     SourceDeathRow,
+    ...     )
     >>> from lp.archivepublisher.diskpool import DiskPool
 
-    >>> disk_pool = DiskPool(pool_path, temp_path, FakeLogger())
-    >>> death_row = DeathRow(ubuntu.main_archive, disk_pool, FakeLogger())
-    >>> death_row.reap(dry_run=True)
+    >>> logger = FakeLogger()
+    >>> disk_pool = DiskPool(pool_path, temp_path, logger)
+    >>> sdr = SourceDeathRow(
+    ...     ubuntu.main_archive, disk_pool, logger, dry_run=True)
+    >>> sdr.run()
+    >>> bdr = BinaryDeathRow(
+    ...     ubuntu.main_archive, disk_pool, logger, dry_run=True)
+    >>> bdr.run()
     DEBUG 0 Sources
     DEBUG 0 Binaries
     INFO Removing 0 files marked for reaping
@@ -237,9 +245,11 @@
 
 Run DeathRow against the current 'removable' context.
 
-    >>> disk_pool = DiskPool(pool_path, temp_path, FakeLogger())
-    >>> death_row = DeathRow(ubuntu.main_archive, disk_pool, FakeLogger())
-    >>> death_row.reap()
+    >>> disk_pool = DiskPool(pool_path, temp_path, logger)
+    >>> sdr = SourceDeathRow(ubuntu.main_archive, disk_pool, logger)
+    >>> sdr.run()
+    >>> bdr = BinaryDeathRow(ubuntu.main_archive, disk_pool, logger)
+    >>> bdr.run()
     DEBUG 4 Sources
     DEBUG 3 Binaries
     ...
@@ -295,15 +305,15 @@
 
 The dependent publications were processed as expected; only the one
 with 'scheduleddeletiondate' set to the past was removed, the one with
-future timestamp and the published one were kept. No binary
-publications was removed (see more below).
+future timestamp and the published one were kept. One binary publication
+was removed, but the file remains on disk for the two active publications.
 
     >>> for pub in dependent_records:
     ...     check_removed(pub)
     stuck 666 in hoary          SUPERSEDED True
     stuck 667 in hoary          SUPERSEDED False
     stuck 668 in hoary          PUBLISHED  False
-    stuck-bin 666 in hoary i386 SUPERSEDED False
+    stuck-bin 666 in hoary i386 SUPERSEDED True
     stuck-bin 666 in hoary i386 SUPERSEDED False
     stuck-bin 666 in hoary i386 PUBLISHED  False
 
@@ -333,9 +343,11 @@
 
 Now DeathRow considers 'stuck-bin' publications.
 
-    >>> disk_pool = DiskPool(pool_path, temp_path, FakeLogger())
-    >>> death_row = DeathRow(ubuntu.main_archive, disk_pool, FakeLogger())
-    >>> death_row.reap()
+    >>> disk_pool = DiskPool(pool_path, temp_path, logger)
+    >>> sdr = SourceDeathRow(ubuntu.main_archive, disk_pool, logger)
+    >>> sdr.run()
+    >>> bdr = BinaryDeathRow(ubuntu.main_archive, disk_pool, logger)
+    >>> bdr.run()
     DEBUG 0 Sources
     DEBUG 2 Binaries
     DEBUG Checking stuck-bin_666_i386.deb (21c2e59531c8710156d34a3c30ac81d5)
@@ -346,6 +358,34 @@
     INFO Total bytes freed: 0
     DEBUG Marking 0 condemned packages as removed.
 
+The file is still in the repository, but now two of the three
+publications are removed.
+
+    >>> dependent_binaries = (
+    ...     published_binary,
+    ...     postponed_binary,
+    ...     removed_binary,
+    ... )
+
+    >>> check_pool_files()
+    deleted-bin_666_i386.deb:    REMOVED
+    deleted_666.dsc:             REMOVED
+    obsolete-bin_666_i386.deb:   REMOVED
+    obsolete_666.dsc:            REMOVED
+    stuck-bin_666_i386.deb:      OK
+    shared_1.0.tar.gz:           OK
+    stuck_666.dsc:               REMOVED
+    stuck_667.dsc:               OK
+    stuck_668.dsc:               OK
+    superseded-bin_666_i386.deb: REMOVED
+    superseded_666.dsc:          REMOVED
+
+    >>> for pub in dependent_binaries:
+    ...     check_removed(pub)
+    stuck-bin 666 in hoary i386 SUPERSEDED True
+    stuck-bin 666 in hoary i386 SUPERSEDED False
+    stuck-bin 666 in hoary i386 SUPERSEDED True
+
 After being considered for removal, DeathRow realized that this binary
 could not be removed because there is still a publishing record
 imposing quarantine on it. Once the quarantine is lifted, by setting a
@@ -357,9 +397,11 @@
 That done, the publication and its files are free to be removed in a
 single pass.
 
-    >>> disk_pool = DiskPool(pool_path, temp_path, FakeLogger())
-    >>> death_row = DeathRow(ubuntu.main_archive, disk_pool, FakeLogger())
-    >>> death_row.reap()
+    >>> disk_pool = DiskPool(pool_path, temp_path, logger)
+    >>> sdr = SourceDeathRow(ubuntu.main_archive, disk_pool, logger)
+    >>> sdr.run()
+    >>> bdr = BinaryDeathRow(ubuntu.main_archive, disk_pool, logger)
+    >>> bdr.run()
     DEBUG 0 Sources
     DEBUG 3 Binaries
     DEBUG Checking stuck-bin_666_i386.deb (21c2e59531c8710156d34a3c30ac81d5)
@@ -374,12 +416,6 @@
 
 The file was removed from the repository.
 
-    >>> dependent_binaries = (
-    ...     published_binary,
-    ...     postponed_binary,
-    ...     removed_binary,
-    ... )
-
     >>> check_pool_files()
     deleted-bin_666_i386.deb:    REMOVED
     deleted_666.dsc:             REMOVED

=== modified file 'lib/lp/archivepublisher/tests/test_deathrow.py'
--- lib/lp/archivepublisher/tests/test_deathrow.py	2012-01-01 02:58:52 +0000
+++ lib/lp/archivepublisher/tests/test_deathrow.py	2016-09-24 06:41:02 +0000
@@ -12,17 +12,22 @@
 
 from zope.component import getUtility
 
-from lp.archivepublisher.deathrow import DeathRow
+from lp.archivepublisher.deathrow import (
+    DeathRow,
+    SourceDeathRow,
+    )
 from lp.archivepublisher.diskpool import DiskPool
 from lp.registry.interfaces.distribution import IDistributionSet
+from lp.services.database.constants import UTC_NOW
 from lp.services.log.logger import BufferLogger
+from lp.soyuz.interfaces.archive import ArchivePurpose
 from lp.soyuz.interfaces.component import IComponentSet
 from lp.soyuz.tests.test_publishing import SoyuzTestPublisher
-from lp.testing import TestCase
+from lp.testing import TestCaseWithFactory
 from lp.testing.layers import LaunchpadZopelessLayer
 
 
-class TestDeathRow(TestCase):
+class TestDeathRow(TestCaseWithFactory):
 
     layer = LaunchpadZopelessLayer
 
@@ -143,3 +148,62 @@
 
         self.assertDoesNotExist(main_dsc_path)
         self.assertDoesNotExist(universe_dsc_path)
+
+    def test_calculateReapableFiles_basic(self):
+        # An isolated publication's files are all removable.
+        spph = self.factory.makeSourcePackagePublishingHistory(
+            archive=self.factory.makeArchive(), sourcepackagename=u'something')
+        for fn in ('something_1.orig.tar.gz', 'something_1.dsc'):
+            self.factory.makeSourcePackageReleaseFile(
+                sourcepackagerelease=spph.sourcepackagerelease,
+                library_file=self.factory.makeLibraryFileAlias(
+                    filename=fn))
+
+        dr = SourceDeathRow(spph.archive, DiskPool, BufferLogger())
+        self.assertContentEqual(
+            [(u'main', u'something', u'something_1.orig.tar.gz'),
+             (u'main', u'something', u'something_1.dsc')],
+            dr.calculateReapableFiles([spph]))
+
+    def test_calculateReapableFiles_cross_component(self):
+        # Files can be removed from a component once all relevant
+        # publications in that component are gone.
+        a = self.factory.makeArchive(purpose=ArchivePurpose.PRIMARY)
+        shared_file = self.factory.makeLibraryFileAlias(
+            filename=u'something_1.orig.tar.gz')
+        sources = []
+        for comp in [u'main', u'universe', u'main', u'main']:
+            spph = self.factory.makeSourcePackagePublishingHistory(
+                archive=a, component=comp, sourcepackagename=u'something')
+            spph.sourcepackagerelease.addFile(shared_file)
+            sources.append(spph)
+
+        log = BufferLogger()
+        dr = SourceDeathRow(a, DiskPool, log)
+
+        # Processing all four pubs simultaneously permits removal of all
+        # their files.
+        self.assertContentEqual(
+            [(u'main', u'something', u'something_1.orig.tar.gz'),
+             (u'universe', u'something', u'something_1.orig.tar.gz')],
+            dr.calculateReapableFiles(sources))
+
+        # When processing just the first two sources, only the universe
+        # file can be removed. The main file is kept alive by the main
+        # publications in the next batch.
+        self.assertContentEqual(
+            [(u'universe', u'something', u'something_1.orig.tar.gz')],
+            dr.calculateReapableFiles(sources[:2]))
+
+        # The second batch (two main publications) can remove nothing,
+        # since all files are kept alive by the main pub in the first
+        # batch.
+        self.assertContentEqual([], dr.calculateReapableFiles(sources[2:]))
+
+        # With the first batch removed, the second batch can eliminate
+        # the main file.
+        for pub in sources[:2]:
+            pub.dateremoved = UTC_NOW
+        self.assertContentEqual(
+            [(u'main', u'something', u'something_1.orig.tar.gz')],
+            dr.calculateReapableFiles(sources[2:]))

=== modified file 'lib/lp/services/database/stormexpr.py'
--- lib/lp/services/database/stormexpr.py	2016-06-10 22:02:37 +0000
+++ lib/lp/services/database/stormexpr.py	2016-09-24 06:41:02 +0000
@@ -104,8 +104,10 @@
         "%s::%s" % (compile(value, state), type)
         for value, type in zip(expr.values[0], col_types))
     rows = [first_row] + [compile(value, state) for value in expr.values[1:]]
-    return "(VALUES (%s)) AS %s(%s)" % (
-        "), (".join(rows), expr.name, ', '.join(col_names))
+    s = "(VALUES (%s))" % "), (".join(rows)
+    if expr.name is not None:
+        s += " AS %s(%s)" % (expr.name, ', '.join(col_names))
+    return s
 
 
 class ColumnSelect(Expr):

=== modified file 'lib/lp/soyuz/interfaces/publishing.py'
--- lib/lp/soyuz/interfaces/publishing.py	2015-11-08 01:09:46 +0000
+++ lib/lp/soyuz/interfaces/publishing.py	2016-09-24 06:41:02 +0000
@@ -290,6 +290,7 @@
             required=False, readonly=False,
             ),
         exported_as="distro_series")
+    componentID = Attribute("DB ID for the component")
     component = Int(
             title=_('The component being published into'),
             required=False, readonly=False,
@@ -680,6 +681,7 @@
             ),
         exported_as="distro_arch_series")
     distroseries = Attribute("The distroseries being published into")
+    componentID = Attribute("DB ID for the component")
     component = Int(
             title=_('The component being published into'),
             required=False, readonly=False,