Merge ~wgrant/launchpad:archive-file-history-refactor into launchpad:master

[William Grant]

Unmerged commits

1bdfbaa... by William Grant on 2020-09-16

Experimental _updateByHash refactoring.

afbab5f... by William Grant on 2020-09-16

Factor _getCurrentFiles out of _updateByHash.

178ada8... by Colin Watson on 2018-04-21

Turn ArchiveFile into a history table

This adds date_created and date_superseded columns. Adjust the
publisher to match.

LP: #1765933

diff --git a/lib/lp/archivepublisher/publishing.py b/lib/lp/archivepublisher/publishing.py
index 4735b89..56ef070 100644
--- a/lib/lp/archivepublisher/publishing.py
+++ b/lib/lp/archivepublisher/publishing.py
@@ -75,6 +75,7 @@ from lp.registry.model.distroseries import DistroSeries
 from lp.services.compat import lzma
 from lp.services.database.constants import UTC_NOW
 from lp.services.database.interfaces import IStore
+from lp.services.database.sqlbase import get_transaction_timestamp
 from lp.services.features import getFeatureFlag
 from lp.services.helpers import filenameToContentType
 from lp.services.librarian.client import LibrarianClient
@@ -95,6 +96,7 @@ from lp.soyuz.interfaces.publishing import (
     active_publishing_status,
     IPublishingSet,
     )
+from lp.soyuz.model.archivefile import ArchiveFile
 from lp.soyuz.model.distroarchseries import DistroArchSeries
 from lp.soyuz.model.publishing import (
     BinaryPackagePublishingHistory,
@@ -1040,15 +1042,13 @@ class Publisher(object):
             return self.distro.displayname
         return "LP-PPA-%s" % get_ppa_reference(self.archive)
 
-    def _updateByHash(self, suite, release_file_name, extra_files):
-        """Update by-hash files for a suite.
+    def _getCurrentFiles(self, suite, release_file_name, extra_files):
+        # Gather information on entries in the current Release file.
+        release_path = os.path.join(
+            self._config.distsroot, suite, release_file_name)
+        with open(release_path) as release_file:
+            release_data = Release(release_file)
 
-        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.
-        """
         extra_data = {}
         for filename, real_filename in extra_files.items():
             hashes = self._readIndexFileHashes(
@@ -1059,22 +1059,9 @@ class Publisher(object):
                 extra_data.setdefault(archive_hash.apt_name, []).append(
                     hashes[archive_hash.deb822_name])
 
-        release_path = os.path.join(
-            self._config.distsroot, suite, release_file_name)
-        with open(release_path) as release_file:
-            release_data = Release(release_file)
-        archive_file_set = getUtility(IArchiveFileSet)
-        by_hashes = ByHashes(self._config.distsroot, self.log)
         suite_dir = os.path.relpath(
             os.path.join(self._config.distsroot, suite),
             self._config.distsroot)
-        container = "release:%s" % suite
-
-        def strip_dists(path):
-            assert path.startswith("dists/")
-            return path[len("dists/"):]
-
-        # Gather information on entries in the current Release file.
         current_files = {}
         for current_entry in (
                 release_data["SHA256"] + extra_data.get("SHA256", [])):
@@ -1083,102 +1070,112 @@ class Publisher(object):
             real_path = os.path.join(suite_dir, real_name)
             current_files[path] = (
                 int(current_entry["size"]), current_entry["sha256"], real_path)
+        return current_files
+
+    def _updateByHash(self, suite, release_file_name, extra_files):
+        """Update by-hash files for a suite.
 
-        # Gather information on entries currently in the database.  Ensure
-        # that we know about all the relevant by-hash directory trees before
-        # doing any removals so that we can prune them properly later, and
-        # work out which condemned files should be reprieved due to the
-        # paths in question having their previous content again.
-        reprieved_files = defaultdict(list)
-        uncondemned_files = set()
+        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)
+        container = "release:%s" % suite
+
+        by_hashes = ByHashes(self._config.distsroot, self.log)
+        existing_lfa_cache = {}
+        existing_live_files = {}
+        reapable_archivefiles = set()
+
+        def strip_dists(path):
+            assert path.startswith("dists/")
+            return path[len("dists/"):]
+
+        # Record all files from the database.
+        db_now = get_transaction_timestamp(IStore(ArchiveFile))
         for db_file in archive_file_set.getByArchive(
                 self.archive, container=container, eager_load=True):
+            file_key = (
+                strip_dists(db_file.path), db_file.library_file.content.sha256)
+            existing_lfa_cache[file_key] = db_file.library_file
+            # Ensure any subdirectories are registered early on, in case we're
+            # about to delete the only file and need to know to prune it.
             by_hashes.registerChild(os.path.dirname(strip_dists(db_file.path)))
-            file_key = (db_file.path, db_file.library_file.content.sha256)
-            if db_file.scheduled_deletion_date is None:
-                uncondemned_files.add(file_key)
-            else:
-                stripped_path = strip_dists(db_file.path)
-                if stripped_path in current_files:
-                    current_sha256 = current_files[stripped_path][1]
-                    if db_file.library_file.content.sha256 == current_sha256:
-                        reprieved_files[file_key].append(db_file)
-
-        # We may already have uncondemned entries with the same path and
-        # content as condemned entries that we were about to reprieve; if
-        # so, there's no need to reprieve them.
-        for file_key in uncondemned_files:
-            reprieved_files.pop(file_key, None)
-
-        # Make sure nothing in the current Release file is condemned.
-        if reprieved_files:
-            reprieved_files_flat = set(
-                chain.from_iterable(reprieved_files.values()))
-            archive_file_set.unscheduleDeletion(reprieved_files_flat)
-            for db_file in reprieved_files_flat:
-                self.log.debug(
-                    "by-hash: Unscheduled %s for %s in %s for deletion" % (
-                        db_file.library_file.content.sha256, db_file.path,
-                        db_file.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 container, path, sha256 in archive_file_set.reap(
-                self.archive, container=container):
-            if (path, sha256) not in uncondemned_files:
-                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(strip_dists(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:
-                stripped_path = strip_dists(db_file.path)
-                if stripped_path in current_files:
-                    current_sha256 = current_files[stripped_path][1]
-                else:
-                    current_sha256 = None
-                if db_file.library_file.content.sha256 != current_sha256:
-                    condemned_files.add(db_file)
-        if condemned_files:
-            archive_file_set.scheduleDeletion(
-                condemned_files, timedelta(days=BY_HASH_STAY_OF_EXECUTION))
-            for db_file in condemned_files:
-                self.log.debug(
-                    "by-hash: Scheduled %s for %s in %s for deletion" % (
-                        db_file.library_file.content.sha256, db_file.path,
-                        db_file.container))
+            if db_file.date_superseded is None:
+                # XXX: Should be a DB constraint.
+                assert file_key not in existing_live_files
+                existing_live_files[file_key] = db_file
+            if (db_file.scheduled_deletion_date is not None
+                    and db_file.scheduled_deletion_date < db_now):
+                # File has expired. Mark it for reaping.
+                reapable_archivefiles.add(db_file)
+            else:
+                # File should still be on disk.
+                by_hashes.add(strip_dists(db_file.path), db_file.library_file)
 
         # 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.
+        new_live_files = set()
+        current_files = self._getCurrentFiles(
+            suite, release_file_name, extra_files)
         for path, (size, sha256, real_path) in current_files.items():
+            file_key = (path, sha256)
+            new_live_files.add(file_key)
             full_path = os.path.join(self._config.distsroot, real_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(
+            if not os.path.exists(full_path):
+                # XXX: Should warn?
+                continue
+            # Ensure there's a current ArchiveFile row.
+            if file_key not in existing_live_files:
+                existing_lfa = existing_lfa_cache.get(file_key)
+                if existing_lfa:
+                    # XXX: Untested?
+                    db_file = archive_file_set.new(
                         self.archive, container, os.path.join("dists", path),
-                        fileobj, size, filenameToContentType(path))
+                        existing_lfa)
+                else:
+                    with open(full_path, "rb") as fileobj:
+                        db_file = archive_file_set.newFromFile(
+                            self.archive, container,
+                            os.path.join("dists", path), fileobj, size,
+                            filenameToContentType(path))
+            # And ensure the by-hash links exist on disk.
+            # XXX: Not sure known is pulling its weight, just avoiding a stat.
+            if not by_hashes.known(path, "SHA256", sha256):
                 by_hashes.add(
                     path, db_file.library_file, copy_from_path=real_path)
 
-        # Finally, remove any files from disk that aren't recorded in the
-        # database and aren't active.
+        # Schedule the deletion of any ArchiveFiles which are current in the DB
+        # but weren't current in the archive this round.
+        # XXX: This order would violate a unique constraint.
+        condemned_files = [
+            af for key, af in existing_live_files.items()
+            if key not in new_live_files]
+        if condemned_files:
+            archive_file_set.scheduleDeletion(
+                condemned_files, timedelta(days=BY_HASH_STAY_OF_EXECUTION))
+            for db_file in condemned_files:
+                self.log.debug(
+                    "by-hash: Scheduled %s for %s in %s for deletion" % (
+                        db_file.library_file.content.sha256, db_file.path,
+                        db_file.container))
+
+        # Remove any files from disk that aren't recorded in the database.
         by_hashes.prune()
 
+        # And remove expired DB rows now that we've pruned them and their dirs
+        # from disk.
+        # XXX: Probably should have a model method. reap doesn't quite work,
+        # since I'd prefer to know it was the same set we just considered.
+        for db_file in reapable_archivefiles:
+            IStore(ArchiveFile).remove(db_file)
+
     def _writeReleaseFile(self, suite, release_data):
         """Write a Release file to the archive (as Release.new).
 
diff --git a/lib/lp/archivepublisher/tests/test_publisher.py b/lib/lp/archivepublisher/tests/test_publisher.py
index 5308e12..c1b6daf 100644
--- a/lib/lp/archivepublisher/tests/test_publisher.py
+++ b/lib/lp/archivepublisher/tests/test_publisher.py
@@ -2626,6 +2626,9 @@ class TestUpdateByHash(TestPublisherBase):
         mocked_datetime.utcnow = lambda: self.times[-1].replace(tzinfo=None)
         self.useFixture(MonkeyPatch(
             'lp.soyuz.model.archivefile._now', lambda: self.times[-1]))
+        self.useFixture(MonkeyPatch(
+            'lp.archivepublisher.publishing.get_transaction_timestamp',
+            lambda _: self.times[-1]))
 
     def advanceTime(self, delta=None, absolute=None):
         if delta is not None: