Make sure the manifest builder only deletes temporary storage rows it created

We were accidentally deleting other image storage rows in the repository, which is really, really bad

data/registry_model/manifestbuilder.py

@@ -7,13 +7,15 @@ from collections import namedtuple
 from flask import session
 
 from data import model
-from data.database import db_transaction
+from data.database import db_transaction, ImageStorage
 from data.registry_model import registry_model
+from image.docker.schema2 import EMPTY_LAYER_BLOB_DIGEST
 
 logger = logging.getLogger(__name__)
 
 ManifestLayer = namedtuple('ManifestLayer', ['layer_id', 'v1_metadata_string', 'db_id'])
-_BuilderState = namedtuple('_BuilderState', ['builder_id', 'images', 'tags', 'checksums'])
+_BuilderState = namedtuple('_BuilderState', ['builder_id', 'images', 'tags', 'checksums',
+                                             'temp_storages'])
 
 _SESSION_KEY = '__manifestbuilder'
 
@@ -23,7 +25,7 @@ def create_manifest_builder(repository_ref, storage):
       and returns it. Returns None if the builder could not be constructed.
   """
   builder_id = str(uuid.uuid4())
-  builder = _ManifestBuilder(repository_ref, _BuilderState(builder_id, {}, {}, {}), storage)
+  builder = _ManifestBuilder(repository_ref, _BuilderState(builder_id, {}, {}, {}, []), storage)
   builder._save_to_session()
   return builder
 
@@ -111,10 +113,6 @@ class _ManifestBuilder(object):
                                                       location_name)
       model.tag.create_temporary_hidden_tag(repository, created, temp_tag_expiration)
 
-      # Mark the image as uploading.
-      created.storage.uploading = True
-      created.storage.save()
-
       # Save its V1 metadata.
       command_list = v1_metadata.get('container_config', {}).get('Cmd', None)
       command = json.dumps(command_list) if command_list else None
@@ -155,7 +153,9 @@ class _ManifestBuilder(object):
       existing_storage = repo_image.storage
       repo_image.storage = blob._db_id
       repo_image.save()
-      existing_storage.delete_instance(recursive=True)
+
+      if existing_storage.uploading:
+        self._builder_state.temp_storages.append(existing_storage.id)
 
     self._builder_state.checksums[layer.layer_id] = computed_checksums
     self._save_to_session()
@@ -198,6 +198,15 @@ class _ManifestBuilder(object):
         and it is expected manifest builders will eventually time out if unused for an
         extended period of time.
     """
+    temp_storages = self._builder_state.temp_storages
+    for storage_id in temp_storages:
+      try:
+        storage = ImageStorage.get(id=storage_id)
+        if storage.uploading and storage.content_checksum != EMPTY_LAYER_BLOB_DIGEST:
+          storage.delete_instance()
+      except ImageStorage.DoesNotExist:
+        pass
+
     session.pop(_SESSION_KEY, None)
 
   def _save_to_session(self):