Merge lp:~cjwatson/launchpad/archive-index-by-hash into lp:launchpad

[Colin Watson]

Add files indexed by Release to the librarian and to ArchiveFile. Publish them in by-hash directories, keeping old versions for a day.

Add files indexed by Release to the librarian and to ArchiveFile. Publish them in by-hash directories, keeping old versions for a day.

DistroSeries.publish_by_hash is useful so that we only do this for series with a version of apt that can make use of it, but it also serves as a circuit breaker in case something goes wrong.

[William Grant (wgrant)]

review: Needs Fixing (code)

[William Grant (wgrant)]

review: Needs Fixing (code)

[Colin Watson (cjwatson)]

Should be worth another look now.

[William Grant (wgrant)]

review: Approve (code)

=== modified file 'lib/lp/archivepublisher/model/ftparchive.py'
--- lib/lp/archivepublisher/model/ftparchive.py	2016-02-09 15:51:19 +0000
+++ lib/lp/archivepublisher/model/ftparchive.py	2016-04-02 00:45:52 +0000
@@ -54,10 +54,14 @@
     """Ensure that the path exists and is an empty directory."""
     if os.path.isdir(path):
         for name in os.listdir(path):
+            if name == "by-hash":
+                # Ignore existing by-hash directories; they will be cleaned
+                # up to match the rest of the directory tree later.
+                continue
             child_path = os.path.join(path, name)
             # Directories containing index files should never have
-            # subdirectories.  Guard against expensive mistakes by not
-            # recursing here.
+            # subdirectories other than by-hash.  Guard against expensive
+            # mistakes by not recursing here.
             os.unlink(child_path)
     else:
         os.makedirs(path, 0o755)

=== modified file 'lib/lp/archivepublisher/publishing.py'
--- lib/lp/archivepublisher/publishing.py	2016-03-30 09:17:31 +0000
+++ lib/lp/archivepublisher/publishing.py	2016-04-02 00:45:52 +0000
@@ -12,7 +12,11 @@
 __metaclass__ = type
 
 import bz2
-from datetime import datetime
+from collections import defaultdict
+from datetime import (
+    datetime,
+    timedelta,
+    )
 import errno
 import gzip
 import hashlib
@@ -31,6 +35,11 @@
     )
 from storm.expr import Desc
 from zope.component import getUtility
+from zope.interface import (
+    Attribute,
+    implementer,
+    Interface,
+    )
 
 from lp.app.interfaces.launchpad import ILaunchpadCelebrities
 from lp.archivepublisher import HARDCODED_COMPONENT_ORDER
@@ -64,8 +73,12 @@
 from lp.services.database.constants import UTC_NOW
 from lp.services.database.interfaces import IStore
 from lp.services.features import getFeatureFlag
+from lp.services.helpers import filenameToContentType
 from lp.services.librarian.client import LibrarianClient
-from lp.services.osutils import open_for_writing
+from lp.services.osutils import (
+    ensure_directory_exists,
+    open_for_writing,
+    )
 from lp.services.utils import file_exists
 from lp.soyuz.enums import (
     ArchivePurpose,
@@ -74,6 +87,7 @@
     PackagePublishingStatus,
     )
 from lp.soyuz.interfaces.archive import NoSuchPPA
+from lp.soyuz.interfaces.archivefile import IArchiveFileSet
 from lp.soyuz.interfaces.publishing import (
     active_publishing_status,
     IPublishingSet,
@@ -96,6 +110,10 @@
     }
 
 
+# Number of days before unreferenced files are removed from by-hash.
+BY_HASH_STAY_OF_EXECUTION = 1
+
+
 def reorder_components(components):
     """Return a list of the components provided.
 
@@ -232,6 +250,152 @@
         return max(len(str(item['size'])) for item in self[key])
 
 
+class IArchiveHash(Interface):
+    """Represents a hash algorithm used for index files."""
+
+    hash_factory = Attribute("A hashlib class suitable for this algorithm.")
+    deb822_name = Attribute(
+        "Algorithm name expected by debian.deb822.Release.")
+    apt_name = Attribute(
+        "Algorithm name used by apt in Release files and by-hash "
+        "subdirectories.")
+    lfc_name = Attribute(
+        "LibraryFileContent attribute name corresponding to this algorithm.")
+
+
+@implementer(IArchiveHash)
+class MD5ArchiveHash:
+    hash_factory = hashlib.md5
+    deb822_name = "md5sum"
+    apt_name = "MD5Sum"
+    lfc_name = "md5"
+
+
+@implementer(IArchiveHash)
+class SHA1ArchiveHash:
+    hash_factory = hashlib.sha1
+    deb822_name = "sha1"
+    apt_name = "SHA1"
+    lfc_name = "sha1"
+
+
+@implementer(IArchiveHash)
+class SHA256ArchiveHash:
+    hash_factory = hashlib.sha256
+    deb822_name = "sha256"
+    apt_name = "SHA256"
+    lfc_name = "sha256"
+
+
+archive_hashes = [
+    MD5ArchiveHash(),
+    SHA1ArchiveHash(),
+    SHA256ArchiveHash(),
+    ]
+
+
+class ByHash:
+    """Represents a single by-hash directory tree."""
+
+    def __init__(self, root, key, log):
+        self.root = root
+        self.path = os.path.join(root, key, "by-hash")
+        self.log = log
+        self.known_digests = defaultdict(lambda: defaultdict(set))
+
+    def add(self, name, lfa, copy_from_path=None):
+        """Ensure that by-hash entries for a single file exist.
+
+        :param name: The name of the file under this directory tree.
+        :param lfa: The `ILibraryFileAlias` to add.
+        :param copy_from_path: If not None, copy file content from here
+            rather than fetching it from the librarian.  This can be used
+            for newly-added files to avoid needing to commit the transaction
+            before calling this method.
+        """
+        for archive_hash in archive_hashes:
+            digest = getattr(lfa.content, archive_hash.lfc_name)
+            digest_path = os.path.join(
+                self.path, archive_hash.apt_name, digest)
+            self.known_digests[archive_hash.apt_name][digest].add(name)
+            if not os.path.exists(digest_path):
+                self.log.debug(
+                    "by-hash: Creating %s for %s" % (digest_path, name))
+                ensure_directory_exists(os.path.dirname(digest_path))
+                if copy_from_path is not None:
+                    os.link(
+                        os.path.join(self.root, copy_from_path), digest_path)
+                else:
+                    with open(digest_path, "wb") as outfile:
+                        lfa.open()
+                        try:
+                            shutil.copyfileobj(lfa, outfile, 4 * 1024 * 1024)
+                        finally:
+                            lfa.close()
+
+    def known(self, name, hashname, digest):
+        """Do we know about a file with this name and digest?"""
+        names = self.known_digests[hashname].get(digest)
+        return names is not None and name in names
+
+    def prune(self):
+        """Remove all by-hash entries that we have not been told to add.
+
+        This also removes the by-hash directory itself if no entries remain.
+        """
+        prune_directory = True
+        for archive_hash in archive_hashes:
+            hash_path = os.path.join(self.path, archive_hash.apt_name)
+            if os.path.exists(hash_path):
+                prune_hash_directory = True
+                for digest in list(os.listdir(hash_path)):
+                    if digest not in self.known_digests[archive_hash.apt_name]:
+                        digest_path = os.path.join(hash_path, digest)
+                        self.log.debug(
+                            "by-hash: Deleting unreferenced %s" % digest_path)
+                        os.unlink(digest_path)
+                    else:
+                        prune_hash_directory = False
+                if prune_hash_directory:
+                    os.rmdir(hash_path)
+                else:
+                    prune_directory = False
+        if prune_directory and os.path.exists(self.path):
+            os.rmdir(self.path)
+
+
+class ByHashes:
+    """Represents all by-hash directory trees in an archive."""
+
+    def __init__(self, root, log):
+        self.root = root
+        self.log = log
+        self.children = {}
+
+    def registerChild(self, dirpath):
+        """Register a single by-hash directory.
+
+        Only directories that have been registered here will be pruned by
+        the `prune` method.
+        """
+        if dirpath not in self.children:
+            self.children[dirpath] = ByHash(self.root, dirpath, self.log)
+        return self.children[dirpath]
+
+    def add(self, path, lfa, copy_from_path=None):
+        dirpath, name = os.path.split(path)
+        self.registerChild(dirpath).add(
+            name, lfa, copy_from_path=copy_from_path)
+
+    def known(self, path, hashname, digest):
+        dirpath, name = os.path.split(path)
+        return self.registerChild(dirpath).known(name, hashname, digest)
+
+    def prune(self):
+        for child in self.children.values():
+            child.prune()
+
+
 class Publisher(object):
     """Publisher is the class used to provide the facility to publish
     files in the pool of a Distribution. The publisher objects will be
@@ -567,10 +731,20 @@
         Otherwise we include only pockets flagged as true in dirty_pockets.
         """
         self.log.debug("* Step D: Generating Release files.")
+
+        archive_file_suites = set()
+        for container in getUtility(IArchiveFileSet).getContainersToReap(
+                self.archive, container_prefix=u"release:"):
+            distroseries, pocket = self.distro.getDistroSeriesAndPocket(
+                container[len(u"release:"):])
+            archive_file_suites.add((distroseries, pocket))
+        self.release_files_needed.update(archive_file_suites)
+
         for distroseries in self.distro:
             for pocket in self.archive.getPockets():
                 if not is_careful:
-                    if not self.isDirty(distroseries, pocket):
+                    if (not self.isDirty(distroseries, pocket) and
+                            (distroseries, pocket) not in archive_file_suites):
                         self.log.debug("Skipping release files for %s/%s" %
                                        (distroseries.name, pocket.name))
                         continue
@@ -811,6 +985,95 @@
             return self.distro.displayname
         return "LP-PPA-%s" % get_ppa_reference(self.archive)
 
+    def _updateByHash(self, suite, release_data):
+        """Update by-hash files for a suite.
+
+        This takes Release file data which references a set of on-disk
+        files, injects any newly-modified files from that set into the
+        librarian and the ArchiveFile table, and updates the on-disk by-hash
+        directories to be in sync with ArchiveFile.  Any on-disk by-hash
+        entries that ceased to be current sufficiently long ago are removed.
+        """
+        archive_file_set = getUtility(IArchiveFileSet)
+        by_hashes = ByHashes(self._config.archiveroot, self.log)
+        suite_dir = os.path.relpath(
+            os.path.join(self._config.distsroot, suite),
+            self._config.archiveroot)
+        container = "release:%s" % suite
+
+        # Gather information on entries in the current Release file, and
+        # make sure nothing there is condemned.
+        current_files = {}
+        current_sha256_checksums = set()
+        for current_entry in release_data["SHA256"]:
+            path = os.path.join(suite_dir, current_entry["name"])
+            current_files[path] = (
+                current_entry["size"], current_entry["sha256"])
+            current_sha256_checksums.add(current_entry["sha256"])
+        for container, path, sha256 in archive_file_set.unscheduleDeletion(
+                self.archive, container=container,
+                sha256_checksums=current_sha256_checksums):
+            self.log.debug(
+                "by-hash: Unscheduled %s for %s in %s for deletion" % (
+                    sha256, path, container))
+
+        # Remove any condemned files from the database whose stay of
+        # execution has elapsed.  We ensure that we know about all the
+        # relevant by-hash directory trees before doing any removals so that
+        # we can prune them properly later.
+        for db_file in archive_file_set.getByArchive(
+                self.archive, container=container):
+            by_hashes.registerChild(os.path.dirname(db_file.path))
+        for container, path, sha256 in archive_file_set.reap(
+                self.archive, container=container):
+            self.log.debug(
+                "by-hash: Deleted %s for %s in %s" % (sha256, path, container))
+
+        # Ensure that all files recorded in the database are in by-hash.
+        db_files = archive_file_set.getByArchive(
+            self.archive, container=container, eager_load=True)
+        for db_file in db_files:
+            by_hashes.add(db_file.path, db_file.library_file)
+
+        # Condemn any database records that do not correspond to current
+        # index files.
+        condemned_files = set()
+        for db_file in db_files:
+            if db_file.scheduled_deletion_date is None:
+                path = db_file.path
+                if path in current_files:
+                    current_sha256 = current_files[path][1]
+                else:
+                    current_sha256 = None
+                if db_file.library_file.content.sha256 != current_sha256:
+                    condemned_files.add(db_file)
+        if condemned_files:
+            for container, path, sha256 in archive_file_set.scheduleDeletion(
+                    condemned_files,
+                    timedelta(days=BY_HASH_STAY_OF_EXECUTION)):
+                self.log.debug(
+                    "by-hash: Scheduled %s for %s in %s for deletion" % (
+                        sha256, path, container))
+
+        # Ensure that all the current index files are in by-hash and have
+        # corresponding database entries.
+        # XXX cjwatson 2016-03-15: This should possibly use bulk creation,
+        # although we can only avoid about a third of the queries since the
+        # librarian client has no bulk upload methods.
+        for path, (size, sha256) in current_files.items():
+            full_path = os.path.join(self._config.archiveroot, path)
+            if (os.path.exists(full_path) and
+                    not by_hashes.known(path, "SHA256", sha256)):
+                with open(full_path, "rb") as fileobj:
+                    db_file = archive_file_set.newFromFile(
+                        self.archive, container, path, fileobj,
+                        size, filenameToContentType(path))
+                by_hashes.add(path, db_file.library_file, copy_from_path=path)
+
+        # Finally, remove any files from disk that aren't recorded in the
+        # database and aren't active.
+        by_hashes.prune()
+
     def _writeReleaseFile(self, suite, release_data):
         """Write a Release file to the archive.
 
@@ -919,9 +1182,14 @@
             hashes = self._readIndexFileHashes(suite, filename)
             if hashes is None:
                 continue
-            release_file.setdefault("MD5Sum", []).append(hashes["md5sum"])
-            release_file.setdefault("SHA1", []).append(hashes["sha1"])
-            release_file.setdefault("SHA256", []).append(hashes["sha256"])
+            for archive_hash in archive_hashes:
+                release_file.setdefault(archive_hash.apt_name, []).append(
+                    hashes[archive_hash.deb822_name])
+
+        if distroseries.publish_by_hash:
+            self._updateByHash(suite, release_file)
+            if distroseries.advertise_by_hash:
+                release_file["Acquire-By-Hash"] = "yes"
 
         self._writeReleaseFile(suite, release_file)
         core_files.add("Release")
@@ -1041,16 +1309,14 @@
         # Schedule this for inclusion in the Release file.
         all_series_files.add(os.path.join(component, "i18n", "Index"))
 
-    def _readIndexFileHashes(self, distroseries_name, file_name,
-                             subpath=None):
+    def _readIndexFileHashes(self, suite, file_name, subpath=None):
         """Read an index file and return its hashes.
 
-        :param distroseries_name: Distro series name
+        :param suite: Suite name.
         :param file_name: Filename relative to the parent container directory.
-        :param subpath: Optional subpath within the distroseries root.
-            Generated indexes will not include this path. If omitted,
-            filenames are assumed to be relative to the distroseries
-            root.
+        :param subpath: Optional subpath within the suite root.  Generated
+            indexes will not include this path.  If omitted, filenames are
+            assumed to be relative to the suite root.
         :return: A dictionary mapping hash field names to dictionaries of
             their components as defined by debian.deb822.Release (e.g.
             {"md5sum": {"md5sum": ..., "size": ..., "name": ...}}), or None
@@ -1058,8 +1324,7 @@
         """
         open_func = open
         full_name = os.path.join(
-            self._config.distsroot, distroseries_name, subpath or '.',
-            file_name)
+            self._config.distsroot, suite, subpath or '.', file_name)
         if not os.path.exists(full_name):
             if os.path.exists(full_name + '.gz'):
                 open_func = gzip.open
@@ -1075,10 +1340,8 @@
                 return None
 
         hashes = {
-            "md5sum": hashlib.md5(),
-            "sha1": hashlib.sha1(),
-            "sha256": hashlib.sha256(),
-            }
+            archive_hash.deb822_name: archive_hash.hash_factory()
+            for archive_hash in archive_hashes}
         size = 0
         with open_func(full_name) as in_file:
             for chunk in iter(lambda: in_file.read(256 * 1024), ""):

=== modified file 'lib/lp/archivepublisher/tests/test_publisher.py'
--- lib/lp/archivepublisher/tests/test_publisher.py	2016-03-30 09:17:31 +0000
+++ lib/lp/archivepublisher/tests/test_publisher.py	2016-04-02 00:45:52 +0000
@@ -7,9 +7,14 @@
 
 import bz2
 import crypt
+from datetime import (
+    datetime,
+    timedelta,
+    )
 from functools import partial
 import gzip
 import hashlib
+from operator import attrgetter
 import os
 import shutil
 import stat
@@ -22,9 +27,20 @@
     import lzma
 except ImportError:
     from backports import lzma
+import pytz
 from testtools.matchers import (
     ContainsAll,
+    DirContains,
+    Equals,
+    FileContains,
+    Is,
     LessThan,
+    Matcher,
+    MatchesListwise,
+    MatchesSetwise,
+    MatchesStructure,
+    Not,
+    PathExists,
     )
 import transaction
 from zope.component import getUtility
@@ -36,6 +52,8 @@
     IArchiveSigningKey,
     )
 from lp.archivepublisher.publishing import (
+    ByHash,
+    ByHashes,
     getPublisher,
     I18nIndex,
     Publisher,
@@ -51,6 +69,7 @@
 from lp.registry.interfaces.series import SeriesStatus
 from lp.services.config import config
 from lp.services.database.constants import UTC_NOW
+from lp.services.database.sqlbase import flush_database_caches
 from lp.services.features import getFeatureFlag
 from lp.services.features.testing import FeatureFixture
 from lp.services.gpg.interfaces import IGPGHandler
@@ -69,12 +88,16 @@
     PackageUploadStatus,
     )
 from lp.soyuz.interfaces.archive import IArchiveSet
+from lp.soyuz.interfaces.archivefile import IArchiveFileSet
 from lp.soyuz.tests.test_publishing import TestNativePublishingBase
 from lp.testing import TestCaseWithFactory
 from lp.testing.fakemethod import FakeMethod
 from lp.testing.gpgkeys import gpgkeysdir
 from lp.testing.keyserver import KeyServerTac
-from lp.testing.layers import ZopelessDatabaseLayer
+from lp.testing.layers import (
+    LaunchpadZopelessLayer,
+    ZopelessDatabaseLayer,
+    )
 
 
 RELEASE = PackagePublishingPocket.RELEASE
@@ -424,6 +447,226 @@
             'i386', publications[0].distroarchseries.architecturetag)
 
 
+class ByHashHasContents(Matcher):
+    """Matches if a by-hash directory has exactly the specified contents."""
+
+    def __init__(self, contents):
+        self.contents = contents
+
+    def match(self, by_hash_path):
+        mismatch = DirContains(["MD5Sum", "SHA1", "SHA256"]).match(
+            by_hash_path)
+        if mismatch is not None:
+            return mismatch
+        for hashname, hashattr in (
+                ("MD5Sum", "md5"), ("SHA1", "sha1"), ("SHA256", "sha256")):
+            digests = {
+                getattr(hashlib, hashattr)(content).hexdigest(): content
+                for content in self.contents}
+            path = os.path.join(by_hash_path, hashname)
+            mismatch = DirContains(digests.keys()).match(path)
+            if mismatch is not None:
+                return mismatch
+            for digest, content in digests.items():
+                mismatch = FileContains(content).match(
+                    os.path.join(path, digest))
+                if mismatch is not None:
+                    return mismatch
+
+
+class ByHashesHaveContents(Matcher):
+    """Matches if only these by-hash directories exist with proper contents."""
+
+    def __init__(self, path_contents):
+        self.path_contents = path_contents
+
+    def match(self, root):
+        children = set()
+        for dirpath, dirnames, _ in os.walk(root):
+            if "by-hash" in dirnames:
+                children.add(os.path.relpath(dirpath, root))
+        mismatch = MatchesSetwise(
+            *(Equals(path) for path in self.path_contents)).match(children)
+        if mismatch is not None:
+            return mismatch
+        for path, contents in self.path_contents.items():
+            by_hash_path = os.path.join(root, path, "by-hash")
+            mismatch = ByHashHasContents(contents).match(by_hash_path)
+            if mismatch is not None:
+                return mismatch
+
+
+class TestByHash(TestCaseWithFactory):
+    """Unit tests for details of handling a single by-hash directory tree."""
+
+    layer = LaunchpadZopelessLayer
+
+    def test_add(self):
+        root = self.makeTemporaryDirectory()
+        contents = ["abc\n", "def\n"]
+        lfas = [
+            self.factory.makeLibraryFileAlias(content=content)
+            for content in contents]
+        transaction.commit()
+        by_hash = ByHash(root, "dists/foo/main/source", DevNullLogger())
+        for lfa in lfas:
+            by_hash.add("Sources", lfa)
+        by_hash_path = os.path.join(root, "dists/foo/main/source/by-hash")
+        self.assertThat(by_hash_path, ByHashHasContents(contents))
+
+    def test_add_copy_from_path(self):
+        root = self.makeTemporaryDirectory()
+        content = "abc\n"
+        sources_path = "dists/foo/main/source/Sources"
+        with open_for_writing(
+                os.path.join(root, sources_path), "w") as sources:
+            sources.write(content)
+        lfa = self.factory.makeLibraryFileAlias(content=content, db_only=True)
+        by_hash = ByHash(root, "dists/foo/main/source", DevNullLogger())
+        by_hash.add("Sources", lfa, copy_from_path=sources_path)
+        by_hash_path = os.path.join(root, "dists/foo/main/source/by-hash")
+        self.assertThat(by_hash_path, ByHashHasContents([content]))
+
+    def test_add_existing(self):
+        root = self.makeTemporaryDirectory()
+        content = "abc\n"
+        lfa = self.factory.makeLibraryFileAlias(content=content)
+        by_hash_path = os.path.join(root, "dists/foo/main/source/by-hash")
+        for hashname, hashattr in (
+                ("MD5Sum", "md5"), ("SHA1", "sha1"), ("SHA256", "sha256")):
+            digest = getattr(hashlib, hashattr)(content).hexdigest()
+            with open_for_writing(
+                    os.path.join(by_hash_path, hashname, digest), "w") as f:
+                f.write(content)
+        by_hash = ByHash(root, "dists/foo/main/source", DevNullLogger())
+        self.assertThat(by_hash_path, ByHashHasContents([content]))
+        by_hash.add("Sources", lfa)
+        self.assertThat(by_hash_path, ByHashHasContents([content]))
+
+    def test_known(self):
+        root = self.makeTemporaryDirectory()
+        content = "abc\n"
+        with open_for_writing(os.path.join(root, "abc"), "w") as f:
+            f.write(content)
+        lfa = self.factory.makeLibraryFileAlias(content=content, db_only=True)
+        by_hash = ByHash(root, "", DevNullLogger())
+        md5 = hashlib.md5(content).hexdigest()
+        sha1 = hashlib.sha1(content).hexdigest()
+        sha256 = hashlib.sha256(content).hexdigest()
+        self.assertFalse(by_hash.known("abc", "MD5Sum", md5))
+        self.assertFalse(by_hash.known("abc", "SHA1", sha1))
+        self.assertFalse(by_hash.known("abc", "SHA256", sha256))
+        by_hash.add("abc", lfa, copy_from_path="abc")
+        self.assertTrue(by_hash.known("abc", "MD5Sum", md5))
+        self.assertTrue(by_hash.known("abc", "SHA1", sha1))
+        self.assertTrue(by_hash.known("abc", "SHA256", sha256))
+        self.assertFalse(by_hash.known("def", "SHA256", sha256))
+        by_hash.add("def", lfa, copy_from_path="abc")
+        self.assertTrue(by_hash.known("def", "SHA256", sha256))
+
+    def test_prune(self):
+        root = self.makeTemporaryDirectory()
+        content = "abc\n"
+        sources_path = "dists/foo/main/source/Sources"
+        with open_for_writing(os.path.join(root, sources_path), "w") as f:
+            f.write(content)
+        lfa = self.factory.makeLibraryFileAlias(content=content, db_only=True)
+        by_hash = ByHash(root, "dists/foo/main/source", DevNullLogger())
+        by_hash.add("Sources", lfa, copy_from_path=sources_path)
+        by_hash_path = os.path.join(root, "dists/foo/main/source/by-hash")
+        with open_for_writing(os.path.join(by_hash_path, "MD5Sum/0"), "w"):
+            pass
+        self.assertThat(by_hash_path, Not(ByHashHasContents([content])))
+        by_hash.prune()
+        self.assertThat(by_hash_path, ByHashHasContents([content]))
+
+    def test_prune_empty(self):
+        root = self.makeTemporaryDirectory()
+        by_hash = ByHash(root, "dists/foo/main/source", DevNullLogger())
+        by_hash_path = os.path.join(root, "dists/foo/main/source/by-hash")
+        with open_for_writing(os.path.join(by_hash_path, "MD5Sum/0"), "w"):
+            pass
+        self.assertThat(by_hash_path, PathExists())
+        by_hash.prune()
+        self.assertThat(by_hash_path, Not(PathExists()))
+
+
+class TestByHashes(TestCaseWithFactory):
+    """Unit tests for details of handling a set of by-hash directory trees."""
+
+    layer = LaunchpadZopelessLayer
+
+    def test_add(self):
+        root = self.makeTemporaryDirectory()
+        self.assertThat(root, ByHashesHaveContents({}))
+        path_contents = {
+            "dists/foo/main/source": {"Sources": "abc\n"},
+            "dists/foo/main/binary-amd64": {
+                "Packages.gz": "def\n", "Packages.xz": "ghi\n"},
+            }
+        by_hashes = ByHashes(root, DevNullLogger())
+        for dirpath, contents in path_contents.items():
+            for name, content in contents.items():
+                path = os.path.join(dirpath, name)
+                with open_for_writing(os.path.join(root, path), "w") as f:
+                    f.write(content)
+                lfa = self.factory.makeLibraryFileAlias(
+                    content=content, db_only=True)
+                by_hashes.add(path, lfa, copy_from_path=path)
+        self.assertThat(root, ByHashesHaveContents({
+            path: contents.values()
+            for path, contents in path_contents.items()}))
+
+    def test_known(self):
+        root = self.makeTemporaryDirectory()
+        content = "abc\n"
+        sources_path = "dists/foo/main/source/Sources"
+        with open_for_writing(os.path.join(root, sources_path), "w") as f:
+            f.write(content)
+        lfa = self.factory.makeLibraryFileAlias(content=content, db_only=True)
+        by_hashes = ByHashes(root, DevNullLogger())
+        md5 = hashlib.md5(content).hexdigest()
+        sha1 = hashlib.sha1(content).hexdigest()
+        sha256 = hashlib.sha256(content).hexdigest()
+        self.assertFalse(by_hashes.known(sources_path, "MD5Sum", md5))
+        self.assertFalse(by_hashes.known(sources_path, "SHA1", sha1))
+        self.assertFalse(by_hashes.known(sources_path, "SHA256", sha256))
+        by_hashes.add(sources_path, lfa, copy_from_path=sources_path)
+        self.assertTrue(by_hashes.known(sources_path, "MD5Sum", md5))
+        self.assertTrue(by_hashes.known(sources_path, "SHA1", sha1))
+        self.assertTrue(by_hashes.known(sources_path, "SHA256", sha256))
+
+    def test_prune(self):
+        root = self.makeTemporaryDirectory()
+        path_contents = {
+            "dists/foo/main/source": {"Sources": "abc\n"},
+            "dists/foo/main/binary-amd64": {
+                "Packages.gz": "def\n", "Packages.xz": "ghi\n"},
+            }
+        by_hashes = ByHashes(root, DevNullLogger())
+        for dirpath, contents in path_contents.items():
+            for name, content in contents.items():
+                path = os.path.join(dirpath, name)
+                with open_for_writing(os.path.join(root, path), "w") as f:
+                    f.write(content)
+                lfa = self.factory.makeLibraryFileAlias(
+                    content=content, db_only=True)
+                by_hashes.add(path, lfa, copy_from_path=path)
+        strays = [
+            "dists/foo/main/source/by-hash/MD5Sum/0",
+            "dists/foo/main/binary-amd64/by-hash/MD5Sum/0",
+            ]
+        for stray in strays:
+            with open_for_writing(os.path.join(root, stray), "w"):
+                pass
+        matcher = ByHashesHaveContents({
+            path: contents.values()
+            for path, contents in path_contents.items()})
+        self.assertThat(root, Not(matcher))
+        by_hashes.prune()
+        self.assertThat(root, matcher)
+
+
 class TestPublisher(TestPublisherBase):
     """Testing `Publisher` behaviour."""
 
@@ -1018,6 +1261,22 @@
         self.assertEqual(
             1 + old_num_pending_archives, new_num_pending_archives)
 
+    def testPendingArchiveWithReapableFiles(self):
+        # getPendingPublicationPPAs returns archives that have reapable
+        # ArchiveFiles.
+        ubuntu = getUtility(IDistributionSet)['ubuntu']
+        archive = self.factory.makeArchive()
+        self.assertNotIn(archive, ubuntu.getPendingPublicationPPAs())
+        archive_file = self.factory.makeArchiveFile(archive=archive)
+        self.assertNotIn(archive, ubuntu.getPendingPublicationPPAs())
+        now = datetime.now(pytz.UTC)
+        removeSecurityProxy(archive_file).scheduled_deletion_date = (
+            now + timedelta(hours=12))
+        self.assertNotIn(archive, ubuntu.getPendingPublicationPPAs())
+        removeSecurityProxy(archive_file).scheduled_deletion_date = (
+            now - timedelta(hours=12))
+        self.assertIn(archive, ubuntu.getPendingPublicationPPAs())
+
     def _checkCompressedFiles(self, archive_publisher, base_file_path,
                               suffixes):
         """Assert that the various compressed versions of a file are equal.
@@ -1930,6 +2189,345 @@
             'Release')
         self.assertTrue(file_exists(source_release))
 
+    def testUpdateByHashDisabled(self):
+        # The publisher does not create by-hash directories if it is
+        # disabled in the series configuration.
+        self.assertFalse(self.breezy_autotest.publish_by_hash)
+        self.assertFalse(self.breezy_autotest.advertise_by_hash)
+        publisher = Publisher(
+            self.logger, self.config, self.disk_pool,
+            self.ubuntutest.main_archive)
+
+        self.getPubSource(filecontent='Source: foo\n')
+
+        publisher.A_publish(False)
+        publisher.C_doFTPArchive(False)
+        publisher.D_writeReleaseFiles(False)
+
+        suite_path = partial(
+            os.path.join, self.config.distsroot, 'breezy-autotest')
+        self.assertThat(
+            suite_path('main', 'source', 'by-hash'), Not(PathExists()))
+        release = self.parseRelease(suite_path('Release'))
+        self.assertNotIn('Acquire-By-Hash', release)
+
+    def testUpdateByHashUnadvertised(self):
+        # If the series configuration sets publish_by_hash but not
+        # advertise_by_hash, then by-hash directories are created but not
+        # advertised in Release.  This is useful for testing.
+        self.breezy_autotest.publish_by_hash = True
+        self.assertFalse(self.breezy_autotest.advertise_by_hash)
+        publisher = Publisher(
+            self.logger, self.config, self.disk_pool,
+            self.ubuntutest.main_archive)
+
+        self.getPubSource(filecontent='Source: foo\n')
+
+        publisher.A_publish(False)
+        publisher.C_doFTPArchive(False)
+        publisher.D_writeReleaseFiles(False)
+
+        suite_path = partial(
+            os.path.join, self.config.distsroot, 'breezy-autotest')
+        self.assertThat(suite_path('main', 'source', 'by-hash'), PathExists())
+        release = self.parseRelease(suite_path('Release'))
+        self.assertNotIn('Acquire-By-Hash', release)
+
+    def testUpdateByHashInitial(self):
+        # An initial publisher run populates by-hash directories and leaves
+        # no archive files scheduled for deletion.
+        self.breezy_autotest.publish_by_hash = True
+        self.breezy_autotest.advertise_by_hash = True
+        publisher = Publisher(
+            self.logger, self.config, self.disk_pool,
+            self.ubuntutest.main_archive)
+
+        self.getPubSource(filecontent='Source: foo\n')
+
+        publisher.A_publish(False)
+        publisher.C_doFTPArchive(False)
+        publisher.D_writeReleaseFiles(False)
+        flush_database_caches()
+
+        suite_path = partial(
+            os.path.join, self.config.distsroot, 'breezy-autotest')
+        contents = set()
+        for name in ('Release', 'Sources.gz', 'Sources.bz2'):
+            with open(suite_path('main', 'source', name), 'rb') as f:
+                contents.add(f.read())
+
+        self.assertThat(
+            suite_path('main', 'source', 'by-hash'),
+            ByHashHasContents(contents))
+
+        archive_files = getUtility(IArchiveFileSet).getByArchive(
+            self.ubuntutest.main_archive)
+        self.assertNotEqual([], archive_files)
+        self.assertEqual([], [
+            archive_file for archive_file in archive_files
+            if archive_file.scheduled_deletion_date is not None])
+
+    def testUpdateByHashSubsequent(self):
+        # A subsequent publisher run updates by-hash directories where
+        # necessary, and marks inactive index files for later deletion.
+        self.breezy_autotest.publish_by_hash = True
+        self.breezy_autotest.advertise_by_hash = True
+        publisher = Publisher(
+            self.logger, self.config, self.disk_pool,
+            self.ubuntutest.main_archive)
+
+        self.getPubSource(filecontent='Source: foo\n')
+
+        publisher.A_publish(False)
+        publisher.C_doFTPArchive(False)
+        publisher.D_writeReleaseFiles(False)
+
+        suite_path = partial(
+            os.path.join, self.config.distsroot, 'breezy-autotest')
+        main_contents = set()
+        universe_contents = set()
+        for name in ('Release', 'Sources.gz', 'Sources.bz2'):
+            with open(suite_path('main', 'source', name), 'rb') as f:
+                main_contents.add(f.read())
+            with open(suite_path('universe', 'source', name), 'rb') as f:
+                universe_contents.add(f.read())
+
+        self.getPubSource(sourcename='baz', filecontent='Source: baz\n')
+
+        publisher.A_publish(False)
+        publisher.C_doFTPArchive(False)
+        publisher.D_writeReleaseFiles(False)
+        flush_database_caches()
+
+        for name in ('Release', 'Sources.gz', 'Sources.bz2'):
+            with open(suite_path('main', 'source', name), 'rb') as f:
+                main_contents.add(f.read())
+
+        self.assertThat(
+            suite_path('main', 'source', 'by-hash'),
+            ByHashHasContents(main_contents))
+        self.assertThat(
+            suite_path('universe', 'source', 'by-hash'),
+            ByHashHasContents(universe_contents))
+
+        archive_files = getUtility(IArchiveFileSet).getByArchive(
+            self.ubuntutest.main_archive)
+        self.assertContentEqual(
+            ['dists/breezy-autotest/main/source/Sources.bz2',
+             'dists/breezy-autotest/main/source/Sources.gz'],
+            [archive_file.path for archive_file in archive_files
+             if archive_file.scheduled_deletion_date is not None])
+
+    def testUpdateByHashIdenticalFiles(self):
+        # Multiple identical files in the same directory receive multiple
+        # ArchiveFile rows, even though they share a by-hash entry.
+        self.breezy_autotest.publish_by_hash = True
+        publisher = Publisher(
+            self.logger, self.config, self.disk_pool,
+            self.ubuntutest.main_archive)
+        suite_path = partial(
+            os.path.join, self.config.distsroot, 'breezy-autotest')
+        get_contents_files = lambda: [
+            archive_file
+            for archive_file in getUtility(IArchiveFileSet).getByArchive(
+                self.ubuntutest.main_archive)
+            if archive_file.path.startswith('dists/breezy-autotest/Contents-')]
+
+        # Create the first file.
+        with open_for_writing(suite_path('Contents-i386'), 'w') as f:
+            f.write('A Contents file\n')
+        publisher.markPocketDirty(
+            self.breezy_autotest, PackagePublishingPocket.RELEASE)
+        publisher.A_publish(False)
+        publisher.C_doFTPArchive(False)
+        publisher.D_writeReleaseFiles(False)
+        flush_database_caches()
+        matchers = [
+            MatchesStructure(
+                path=Equals('dists/breezy-autotest/Contents-i386'),
+                scheduled_deletion_date=Is(None))]
+        self.assertThat(get_contents_files(), MatchesSetwise(*matchers))
+        self.assertThat(
+            suite_path('by-hash'), ByHashHasContents(['A Contents file\n']))
+
+        # Add a second identical file.
+        with open_for_writing(suite_path('Contents-hppa'), 'w') as f:
+            f.write('A Contents file\n')
+        publisher.D_writeReleaseFiles(False)
+        flush_database_caches()
+        matchers.append(
+            MatchesStructure(
+                path=Equals('dists/breezy-autotest/Contents-hppa'),
+                scheduled_deletion_date=Is(None)))
+        self.assertThat(get_contents_files(), MatchesSetwise(*matchers))
+        self.assertThat(
+            suite_path('by-hash'), ByHashHasContents(['A Contents file\n']))
+
+        # Delete the first file, but allow it its stay of execution.
+        os.unlink(suite_path('Contents-i386'))
+        publisher.D_writeReleaseFiles(False)
+        flush_database_caches()
+        matchers[0] = matchers[0].update(scheduled_deletion_date=Not(Is(None)))
+        self.assertThat(get_contents_files(), MatchesSetwise(*matchers))
+        self.assertThat(
+            suite_path('by-hash'), ByHashHasContents(['A Contents file\n']))
+
+        # Arrange for the first file to be pruned, and delete the second
+        # file.
+        now = datetime.now(pytz.UTC)
+        i386_file = getUtility(IArchiveFileSet).getByArchive(
+            self.ubuntutest.main_archive,
+            path=u'dists/breezy-autotest/Contents-i386').one()
+        removeSecurityProxy(i386_file).scheduled_deletion_date = (
+            now - timedelta(hours=1))
+        os.unlink(suite_path('Contents-hppa'))
+        publisher.D_writeReleaseFiles(False)
+        flush_database_caches()
+        matchers = [matchers[1].update(scheduled_deletion_date=Not(Is(None)))]
+        self.assertThat(get_contents_files(), MatchesSetwise(*matchers))
+        self.assertThat(
+            suite_path('by-hash'), ByHashHasContents(['A Contents file\n']))
+
+        # Arrange for the second file to be pruned.
+        hppa_file = getUtility(IArchiveFileSet).getByArchive(
+            self.ubuntutest.main_archive,
+            path=u'dists/breezy-autotest/Contents-hppa').one()
+        removeSecurityProxy(hppa_file).scheduled_deletion_date = (
+            now - timedelta(hours=1))
+        publisher.D_writeReleaseFiles(False)
+        flush_database_caches()
+        self.assertContentEqual([], get_contents_files())
+        self.assertThat(suite_path('by-hash'), Not(PathExists()))
+
+    def testUpdateByHashReprieve(self):
+        # If a newly-modified index file is identical to a
+        # previously-condemned one, then it is reprieved and not pruned.
+        self.breezy_autotest.publish_by_hash = True
+        # Enable uncompressed index files to avoid relying on stable output
+        # from compressors in this test.
+        self.breezy_autotest.index_compressors = [
+            IndexCompressionType.UNCOMPRESSED]
+        publisher = Publisher(
+            self.logger, self.config, self.disk_pool,
+            self.ubuntutest.main_archive)
+
+        # Publish empty index files.
+        publisher.markPocketDirty(
+            self.breezy_autotest, PackagePublishingPocket.RELEASE)
+        publisher.A_publish(False)
+        publisher.C_doFTPArchive(False)
+        publisher.D_writeReleaseFiles(False)
+        suite_path = partial(
+            os.path.join, self.config.distsroot, 'breezy-autotest')
+        main_contents = set()
+        for name in ('Release', 'Sources'):
+            with open(suite_path('main', 'source', name), 'rb') as f:
+                main_contents.add(f.read())
+
+        # Add a source package so that Sources is non-empty.
+        pub_source = self.getPubSource(filecontent='Source: foo\n')
+        publisher.A_publish(False)
+        publisher.C_doFTPArchive(False)
+        publisher.D_writeReleaseFiles(False)
+        transaction.commit()
+        with open(suite_path('main', 'source', 'Sources'), 'rb') as f:
+            main_contents.add(f.read())
+        self.assertEqual(3, len(main_contents))
+        self.assertThat(
+            suite_path('main', 'source', 'by-hash'),
+            ByHashHasContents(main_contents))
+
+        # Make the empty Sources file ready to prune.
+        old_archive_files = []
+        for archive_file in getUtility(IArchiveFileSet).getByArchive(
+                self.ubuntutest.main_archive):
+            if ('main/source' in archive_file.path and
+                    archive_file.scheduled_deletion_date is not None):
+                old_archive_files.append(archive_file)
+        self.assertEqual(1, len(old_archive_files))
+        removeSecurityProxy(old_archive_files[0]).scheduled_deletion_date = (
+            datetime.now(pytz.UTC) - timedelta(hours=1))
+
+        # Delete the source package so that Sources is empty again.  The
+        # empty file is reprieved and the non-empty one is condemned.
+        pub_source.requestDeletion(self.ubuntutest.owner)
+        publisher.A_publish(False)
+        publisher.C_doFTPArchive(False)
+        publisher.D_writeReleaseFiles(False)
+        transaction.commit()
+        self.assertThat(
+            suite_path('main', 'source', 'by-hash'),
+            ByHashHasContents(main_contents))
+        archive_files = getUtility(IArchiveFileSet).getByArchive(
+            self.ubuntutest.main_archive,
+            path=u'dists/breezy-autotest/main/source/Sources')
+        self.assertThat(
+            sorted(archive_files, key=attrgetter('id')),
+            MatchesListwise([
+                MatchesStructure(scheduled_deletion_date=Is(None)),
+                MatchesStructure(scheduled_deletion_date=Not(Is(None))),
+                ]))
+
+    def testUpdateByHashPrune(self):
+        # The publisher prunes files from by-hash that were condemned more
+        # than a day ago.
+        self.breezy_autotest.publish_by_hash = True
+        self.breezy_autotest.advertise_by_hash = True
+        publisher = Publisher(
+            self.logger, self.config, self.disk_pool,
+            self.ubuntutest.main_archive)
+
+        suite_path = partial(
+            os.path.join, self.config.distsroot, 'breezy-autotest')
+        main_contents = set()
+        for sourcename in ('foo', 'bar'):
+            self.getPubSource(
+                sourcename=sourcename, filecontent='Source: %s\n' % sourcename)
+            publisher.A_publish(False)
+            publisher.C_doFTPArchive(False)
+            publisher.D_writeReleaseFiles(False)
+            for name in ('Release', 'Sources.gz', 'Sources.bz2'):
+                with open(suite_path('main', 'source', name), 'rb') as f:
+                    main_contents.add(f.read())
+        transaction.commit()
+        # Undo any previous determination that breezy-autotest is dirty, so
+        # that we can use that to check that future runs don't force index
+        # regeneration.
+        publisher.dirty_pockets = set()
+
+        self.assertThat(
+            suite_path('main', 'source', 'by-hash'),
+            ByHashHasContents(main_contents))
+        old_archive_files = []
+        for archive_file in getUtility(IArchiveFileSet).getByArchive(
+                self.ubuntutest.main_archive):
+            if ('main/source' in archive_file.path and
+                    archive_file.scheduled_deletion_date is not None):
+                old_archive_files.append(archive_file)
+        self.assertEqual(2, len(old_archive_files))
+
+        now = datetime.now(pytz.UTC)
+        removeSecurityProxy(old_archive_files[0]).scheduled_deletion_date = (
+            now + timedelta(hours=12))
+        removeSecurityProxy(old_archive_files[1]).scheduled_deletion_date = (
+            now - timedelta(hours=12))
+        old_archive_files[1].library_file.open()
+        try:
+            main_contents.remove(old_archive_files[1].library_file.read())
+        finally:
+            old_archive_files[1].library_file.close()
+        self.assertThat(
+            suite_path('main', 'source', 'by-hash'),
+            Not(ByHashHasContents(main_contents)))
+
+        publisher.A2_markPocketsWithDeletionsDirty()
+        publisher.C_doFTPArchive(False)
+        publisher.D_writeReleaseFiles(False)
+        self.assertEqual(set(), publisher.dirty_pockets)
+        self.assertThat(
+            suite_path('main', 'source', 'by-hash'),
+            ByHashHasContents(main_contents))
+
     def testCreateSeriesAliasesNoAlias(self):
         """createSeriesAliases has nothing to do by default."""
         publisher = Publisher(

=== modified file 'lib/lp/registry/model/distribution.py'
--- lib/lp/registry/model/distribution.py	2015-10-13 13:22:08 +0000
+++ lib/lp/registry/model/distribution.py	2016-04-02 00:45:52 +0000
@@ -1,4 +1,4 @@
-# Copyright 2009-2015 Canonical Ltd.  This software is licensed under the
+# Copyright 2009-2016 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """Database classes for implementing distribution items."""
@@ -1283,10 +1283,22 @@
             bin_query, clauseTables=['BinaryPackagePublishingHistory'],
             orderBy=['archive.id'], distinct=True)
 
+        reapable_af_query = """
+        Archive.purpose = %s AND
+        Archive.distribution = %s AND
+        ArchiveFile.archive = archive.id AND
+        ArchiveFile.scheduled_deletion_date < %s
+        """ % sqlvalues(ArchivePurpose.PPA, self, UTC_NOW)
+
+        reapable_af_archives = Archive.select(
+            reapable_af_query, clauseTables=['ArchiveFile'],
+            orderBy=['archive.id'], distinct=True)
+
         deleting_archives = Archive.selectBy(
             status=ArchiveStatus.DELETING).orderBy(['archive.id'])
 
-        return src_archives.union(bin_archives).union(deleting_archives)
+        return src_archives.union(bin_archives).union(
+            reapable_af_archives).union(deleting_archives)
 
     def getArchiveByComponent(self, component_name):
         """See `IDistribution`."""

=== modified file 'lib/lp/services/helpers.py'
--- lib/lp/services/helpers.py	2014-05-07 15:28:50 +0000
+++ lib/lp/services/helpers.py	2016-04-02 00:45:52 +0000
@@ -10,6 +10,7 @@
 
 __metaclass__ = type
 
+from collections import OrderedDict
 from difflib import unified_diff
 import re
 from StringIO import StringIO
@@ -224,19 +225,37 @@
 
     >>> filenameToContentType('test.tgz')
     'application/octet-stream'
+
+    Build logs
+    >>> filenameToContentType('buildlog.txt.gz')
+    'text/plain'
+
+    Various compressed files
+
+    >>> filenameToContentType('Packages.gz')
+    'application/x-gzip'
+    >>> filenameToContentType('Packages.bz2')
+    'application/x-bzip2'
+    >>> filenameToContentType('Packages.xz')
+    'application/x-xz'
     """
-    ftmap = {".dsc": "text/plain",
-             ".changes": "text/plain",
-             ".deb": "application/x-debian-package",
-             ".udeb": "application/x-debian-package",
-             ".txt": "text/plain",
-             # For the build master logs
-             ".txt.gz": "text/plain",
-             # For live filesystem builds
-             ".manifest": "text/plain",
-             ".manifest-remove": "text/plain",
-             ".size": "text/plain",
-             }
+    ftmap = OrderedDict([
+        (".dsc", "text/plain"),
+        (".changes", "text/plain"),
+        (".deb", "application/x-debian-package"),
+        (".udeb", "application/x-debian-package"),
+        (".txt", "text/plain"),
+        # For the build master logs
+        (".txt.gz", "text/plain"),
+        # For live filesystem builds
+        (".manifest", "text/plain"),
+        (".manifest-remove", "text/plain"),
+        (".size", "text/plain"),
+        # Compressed files
+        (".gz", "application/x-gzip"),
+        (".bz2", "application/x-bzip2"),
+        (".xz", "application/x-xz"),
+        ])
     for ending in ftmap:
         if fname.endswith(ending):
             return ftmap[ending]

=== modified file 'lib/lp/services/librarian/interfaces/__init__.py'
--- lib/lp/services/librarian/interfaces/__init__.py	2016-03-14 16:28:19 +0000
+++ lib/lp/services/librarian/interfaces/__init__.py	2016-04-02 00:45:52 +0000
@@ -155,7 +155,7 @@
 class ILibraryFileAliasSet(Interface):
 
     def create(name, size, file, contentType, expires=None, debugID=None,
-               restricted=False):
+               restricted=False, allow_zero_length=False):
         """Create a file in the Librarian, returning the new alias.
 
         An expiry time of None means the file will never expire until it

=== modified file 'lib/lp/services/librarian/model.py'
--- lib/lp/services/librarian/model.py	2016-03-14 16:28:19 +0000
+++ lib/lp/services/librarian/model.py	2016-04-02 00:45:52 +0000
@@ -244,7 +244,7 @@
     """Create and find LibraryFileAliases."""
 
     def create(self, name, size, file, contentType, expires=None,
-               debugID=None, restricted=False):
+               debugID=None, restricted=False, allow_zero_length=False):
         """See `ILibraryFileAliasSet`"""
         if restricted:
             client = getUtility(IRestrictedLibrarianClient)
@@ -252,7 +252,9 @@
             client = getUtility(ILibrarianClient)
         if '/' in name:
             raise InvalidFilename("Filename cannot contain slashes.")
-        fid = client.addFile(name, size, file, contentType, expires, debugID)
+        fid = client.addFile(
+            name, size, file, contentType, expires=expires, debugID=debugID,
+            allow_zero_length=allow_zero_length)
         lfa = IMasterStore(LibraryFileAlias).find(
             LibraryFileAlias, LibraryFileAlias.id == fid).one()
         assert lfa is not None, "client.addFile didn't!"

=== modified file 'lib/lp/soyuz/interfaces/archivefile.py'
--- lib/lp/soyuz/interfaces/archivefile.py	2016-03-18 15:09:37 +0000
+++ lib/lp/soyuz/interfaces/archivefile.py	2016-04-02 00:45:52 +0000
@@ -79,13 +79,15 @@
         :param content_type: The MIME type of the file.
         """
 
-    def getByArchive(archive, container=None, eager_load=False):
+    def getByArchive(archive, container=None, path=None, eager_load=False):
         """Get files in an archive.
 
         :param archive: Return files in this `IArchive`.
         :param container: Return only files with this container.
+        :param path: Return only files with this path.
         :param eager_load: If True, preload related `LibraryFileAlias` and
             `LibraryFileContent` rows.
+        :return: An iterable of matched files.
         """
 
     def scheduleDeletion(archive_files, stay_of_execution):
@@ -94,6 +96,25 @@
         :param archive_files: The `IArchiveFile`s to schedule for deletion.
         :param stay_of_execution: A `timedelta`; schedule files for deletion
             this amount of time in the future.
+        :return: An iterable of (container, path, sha256) for files that
+            were scheduled for deletion.
+        """
+
+    def unscheduleDeletion(archive, container=None, sha256_checksums=set()):
+        """Unschedule these archive files for deletion.
+
+        This is useful in the case when the new content of a file is
+        identical to a version that was previously condemned.  This method's
+        signature does not match that of `scheduleDeletion`; this is more
+        convenient because in such cases we normally do not yet have
+        `ArchiveFile` rows in hand.
+
+        :param archive: Operate on files in this `IArchive`.
+        :param container: Operate only on files with this container.
+        :param sha256_checksums: Operate only on files with any of these
+            checksums.
+        :return: An iterable of (container, path, sha256) for files that
+            were unscheduled for deletion.
         """
 
     def getContainersToReap(archive, container_prefix=None):
@@ -102,6 +123,7 @@
         :param archive: Return containers in this `IArchive`.
         :param container_prefix: Return only containers that start with this
             prefix.
+        :return: An iterable of matched container names.
         """
 
     def reap(archive, container=None):
@@ -109,4 +131,6 @@
 
         :param archive: Delete files from this `IArchive`.
         :param container: Delete only files with this container.
+        :return: An iterable of (container, path, sha256) for files that
+            were deleted.
         """

=== modified file 'lib/lp/soyuz/model/archivefile.py'
--- lib/lp/soyuz/model/archivefile.py	2016-03-18 15:09:37 +0000
+++ lib/lp/soyuz/model/archivefile.py	2016-04-02 00:45:52 +0000
@@ -14,7 +14,9 @@
 import os.path
 
 import pytz
+from storm.databases.postgres import Returning
 from storm.locals import (
+    And,
     DateTime,
     Int,
     Reference,
@@ -31,6 +33,7 @@
     IMasterStore,
     IStore,
     )
+from lp.services.database.stormexpr import BulkUpdate
 from lp.services.librarian.interfaces import ILibraryFileAliasSet
 from lp.services.librarian.model import (
     LibraryFileAlias,
@@ -89,17 +92,19 @@
                     content_type):
         library_file = getUtility(ILibraryFileAliasSet).create(
             os.path.basename(path), size, fileobj, content_type,
-            restricted=archive.private)
+            restricted=archive.private, allow_zero_length=True)
         return cls.new(archive, container, path, library_file)
 
     @staticmethod
-    def getByArchive(archive, container=None, eager_load=False):
+    def getByArchive(archive, container=None, path=None, eager_load=False):
         """See `IArchiveFileSet`."""
         clauses = [ArchiveFile.archive == archive]
         # XXX cjwatson 2016-03-15: We'll need some more sophisticated way to
         # match containers once we're using them for custom uploads.
         if container is not None:
             clauses.append(ArchiveFile.container == container)
+        if path is not None:
+            clauses.append(ArchiveFile.path == path)
         archive_files = IStore(ArchiveFile).find(ArchiveFile, *clauses)
 
         def eager_load(rows):
@@ -114,11 +119,43 @@
     @staticmethod
     def scheduleDeletion(archive_files, stay_of_execution):
         """See `IArchiveFileSet`."""
-        archive_file_ids = set(
-            archive_file.id for archive_file in archive_files)
-        rows = IMasterStore(ArchiveFile).find(
-            ArchiveFile, ArchiveFile.id.is_in(archive_file_ids))
-        rows.set(scheduled_deletion_date=UTC_NOW + stay_of_execution)
+        clauses = [
+            ArchiveFile.id.is_in(
+                set(archive_file.id for archive_file in archive_files)),
+            ArchiveFile.library_file == LibraryFileAlias.id,
+            LibraryFileAlias.content == LibraryFileContent.id,
+            ]
+        new_date = UTC_NOW + stay_of_execution
+        return_columns = [
+            ArchiveFile.container, ArchiveFile.path, LibraryFileContent.sha256]
+        return list(IMasterStore(ArchiveFile).execute(Returning(
+            BulkUpdate(
+                {ArchiveFile.scheduled_deletion_date: new_date},
+                table=ArchiveFile,
+                values=[LibraryFileAlias, LibraryFileContent],
+                where=And(*clauses)),
+            columns=return_columns)))
+
+    @staticmethod
+    def unscheduleDeletion(archive, container=None, sha256_checksums=set()):
+        """See `IArchiveFileSet`."""
+        clauses = [
+            ArchiveFile.archive == archive,
+            ArchiveFile.library_file == LibraryFileAlias.id,
+            LibraryFileAlias.content == LibraryFileContent.id,
+            LibraryFileContent.sha256.is_in(sha256_checksums),
+            ]
+        if container is not None:
+            clauses.append(ArchiveFile.container == container)
+        return_columns = [
+            ArchiveFile.container, ArchiveFile.path, LibraryFileContent.sha256]
+        return list(IMasterStore(ArchiveFile).execute(Returning(
+            BulkUpdate(
+                {ArchiveFile.scheduled_deletion_date: None},
+                table=ArchiveFile,
+                values=[LibraryFileAlias, LibraryFileContent],
+                where=And(*clauses)),
+            columns=return_columns)))
 
     @staticmethod
     def getContainersToReap(archive, container_prefix=None):
@@ -134,10 +171,25 @@
     @staticmethod
     def reap(archive, container=None):
         """See `IArchiveFileSet`."""
+        # XXX cjwatson 2016-03-30 bug=322972: Requires manual SQL due to
+        # lack of support for DELETE FROM ... USING ... in Storm.
         clauses = [
-            ArchiveFile.archive == archive,
-            ArchiveFile.scheduled_deletion_date < UTC_NOW,
+            "ArchiveFile.archive = ?",
+            "ArchiveFile.scheduled_deletion_date < "
+                "CURRENT_TIMESTAMP AT TIME ZONE 'UTC'",
+            "ArchiveFile.library_file = LibraryFileAlias.id",
+            "LibraryFileAlias.content = LibraryFileContent.id",
             ]
+        values = [archive.id]
         if container is not None:
-            clauses.append(ArchiveFile.container == container)
-        IMasterStore(ArchiveFile).find(ArchiveFile, *clauses).remove()
+            clauses.append("ArchiveFile.container = ?")
+            values.append(container)
+        return list(IMasterStore(ArchiveFile).execute("""
+            DELETE FROM ArchiveFile
+            USING LibraryFileAlias, LibraryFileContent
+            WHERE """ + " AND ".join(clauses) + """
+            RETURNING
+                ArchiveFile.container,
+                ArchiveFile.path,
+                LibraryFileContent.sha256
+            """, values))

=== modified file 'lib/lp/soyuz/tests/test_archivefile.py'
--- lib/lp/soyuz/tests/test_archivefile.py	2016-03-18 15:09:37 +0000
+++ lib/lp/soyuz/tests/test_archivefile.py	2016-04-02 00:45:52 +0000
@@ -19,6 +19,7 @@
 from zope.component import getUtility
 from zope.security.proxy import removeSecurityProxy
 
+from lp.services.database.sqlbase import flush_database_caches
 from lp.services.osutils import open_for_writing
 from lp.soyuz.interfaces.archivefile import IArchiveFileSet
 from lp.testing import TestCaseWithFactory
@@ -75,17 +76,35 @@
         self.assertContentEqual(
             [], archive_file_set.getByArchive(archives[0], container="bar"))
         self.assertContentEqual(
+            [archive_files[1]],
+            archive_file_set.getByArchive(
+                archives[0], path=archive_files[1].path))
+        self.assertContentEqual(
+            [], archive_file_set.getByArchive(archives[0], path="other"))
+        self.assertContentEqual(
             archive_files[2:], archive_file_set.getByArchive(archives[1]))
         self.assertContentEqual(
             [archive_files[3]],
             archive_file_set.getByArchive(archives[1], container="foo"))
         self.assertContentEqual(
             [], archive_file_set.getByArchive(archives[1], container="bar"))
+        self.assertContentEqual(
+            [archive_files[3]],
+            archive_file_set.getByArchive(
+                archives[1], path=archive_files[3].path))
+        self.assertContentEqual(
+            [], archive_file_set.getByArchive(archives[1], path="other"))
 
     def test_scheduleDeletion(self):
         archive_files = [self.factory.makeArchiveFile() for _ in range(3)]
-        getUtility(IArchiveFileSet).scheduleDeletion(
+        expected_rows = [
+            (archive_file.container, archive_file.path,
+             archive_file.library_file.content.sha256)
+            for archive_file in archive_files[:2]]
+        rows = getUtility(IArchiveFileSet).scheduleDeletion(
             archive_files[:2], timedelta(days=1))
+        self.assertContentEqual(expected_rows, rows)
+        flush_database_caches()
         tomorrow = datetime.now(pytz.UTC) + timedelta(days=1)
         # Allow a bit of timing slack for slow tests.
         self.assertThat(
@@ -96,6 +115,34 @@
             LessThan(timedelta(minutes=5)))
         self.assertIsNone(archive_files[2].scheduled_deletion_date)
 
+    def test_unscheduleDeletion(self):
+        archives = [self.factory.makeArchive() for _ in range(2)]
+        lfas = [
+            self.factory.makeLibraryFileAlias(db_only=True) for _ in range(3)]
+        archive_files = []
+        for archive in archives:
+            for container in ("foo", "bar"):
+                archive_files.extend([
+                    self.factory.makeArchiveFile(
+                        archive=archive, container=container, library_file=lfa)
+                    for lfa in lfas])
+        now = datetime.now(pytz.UTC)
+        for archive_file in archive_files:
+            removeSecurityProxy(archive_file).scheduled_deletion_date = now
+        expected_rows = [
+            ("foo", archive_files[0].path, lfas[0].content.sha256),
+            ("foo", archive_files[1].path, lfas[1].content.sha256),
+            ]
+        rows = getUtility(IArchiveFileSet).unscheduleDeletion(
+            archive=archives[0], container="foo",
+            sha256_checksums=[lfas[0].content.sha256, lfas[1].content.sha256])
+        self.assertContentEqual(expected_rows, rows)
+        flush_database_caches()
+        self.assertContentEqual(
+            [archive_files[0], archive_files[1]],
+            [archive_file for archive_file in archive_files
+             if archive_file.scheduled_deletion_date is None])
+
     def test_getContainersToReap(self):
         archive = self.factory.makeArchive()
         archive_files = []
@@ -149,6 +196,11 @@
         removeSecurityProxy(archive_files[4]).scheduled_deletion_date = (
             now - timedelta(days=1))
         archive_file_set = getUtility(IArchiveFileSet)
-        archive_file_set.reap(archive, container="foo")
+        expected_rows = [
+            ("foo", archive_files[0].path,
+             archive_files[0].library_file.content.sha256),
+            ]
+        rows = archive_file_set.reap(archive, container="foo")
+        self.assertContentEqual(expected_rows, rows)
         self.assertContentEqual(
             archive_files[1:4], archive_file_set.getByArchive(archive))