Fix TagManifests with shared digests under the same repository.

TagManifests can (apparently, in very rare scenarios) share manifests with the exact same digests, so we need to support that case in the backfill worker. We also need to remove a unique constraint on the manifest column in the mapping table to support this case.

workers/manifestbackfillworker.py

@@ -5,7 +5,7 @@ from peewee import JOIN, fn, IntegrityError
 
 from app import app
 from data.database import (UseThenDisconnect, TagManifest, TagManifestToManifest, Image,
-                           db_transaction)
+                           Manifest, db_transaction)
 from data.model.image import get_parent_images
 from data.model.tag import populate_manifest
 from data.model.blob import get_repo_blob_by_digest, BlobDoesNotExist
@@ -158,9 +158,22 @@ def backfill_manifest(tag_manifest):
     except TagManifest.DoesNotExist:
       return True
 
-    # Create the new-style rows for the manifest.
-    manifest_row = populate_manifest(tag_manifest.tag.repository, manifest,
-                                     tag_manifest.tag.image, storage_ids)
+    # Ensure it wasn't already created.
+    if lookup_map_row(tag_manifest):
+      return False
+
+    # Check for a pre-existing manifest matching the digest in the repository. This can happen
+    # if we've already created the manifest row (typically for tag reverision).
+    try:
+      manifest_row = Manifest.get(digest=manifest.digest, repository=tag_manifest.tag.repository)
+    except Manifest.DoesNotExist:
+      # Create the new-style rows for the manifest.
+      try:
+        manifest_row = populate_manifest(tag_manifest.tag.repository, manifest,
+                                         tag_manifest.tag.image, storage_ids)
+      except IntegrityError:
+        # Pre-empted.
+        return False
 
     # Create the mapping row. If we find another was created for this tag manifest in the
     # meantime, then we've been preempted.