Merge ~twom/launchpad:oci-registry-upload into launchpad:master

[Tom Wardill]

Upload built images to an OCI registry

[Colin Watson (cjwatson)]

review: Needs Fixing

[Colin Watson (cjwatson)]

Some things to fix, but mostly looks OK now. As you probably know, landing is blocked until https://portal.admin.canonical.com/C125275 is done.

review: Approve

diff --git a/database/schema/security.cfg b/database/schema/security.cfg
index 942c5b4..55226e1 100644
--- a/database/schema/security.cfg
+++ b/database/schema/security.cfg
@@ -999,6 +999,7 @@ public.livefsfile                             = SELECT
 public.ocifile                                = SELECT
 public.ociproject                             = SELECT
 public.ociprojectname                         = SELECT
+public.ocipushrule                            = SELECT
 public.ocirecipe                              = SELECT
 public.ocirecipebuild                         = SELECT, UPDATE
 public.ocirecipebuildjob                      = SELECT, INSERT
@@ -1439,6 +1440,7 @@ public.ocifile                          = SELECT, INSERT
 public.ociproject                       = SELECT
 public.ociprojectname                   = SELECT
 public.ociprojectseries                 = SELECT
+public.ocipushrule                      = SELECT
 public.ocirecipe                        = SELECT, UPDATE
 public.ocirecipebuild                   = SELECT, UPDATE
 public.ocirecipebuildjob                = SELECT, INSERT, UPDATE
@@ -2678,3 +2680,42 @@ public.teammembership                   = SELECT
 public.teamparticipation                = SELECT
 public.webhook                          = SELECT
 public.webhookjob                       = SELECT, INSERT
+
+[oci-build-job]
+type=user
+groups=script
+public.account                          = SELECT
+public.archive                          = SELECT
+public.branch                           = SELECT
+public.builder                          = SELECT
+public.buildfarmjob                     = SELECT, INSERT
+public.buildqueue                       = SELECT, INSERT, UPDATE
+public.distribution                     = SELECT
+public.distroarchseries                 = SELECT
+public.distroseries                     = SELECT
+public.emailaddress                     = SELECT
+public.gitref                           = SELECT
+public.gitrepository                    = SELECT
+public.job                              = SELECT, INSERT, UPDATE
+public.libraryfilealias                 = SELECT
+public.libraryfilecontent               = SELECT
+public.person                           = SELECT
+public.personsettings                   = SELECT
+public.pocketchroot                     = SELECT
+public.processor                        = SELECT
+public.product                          = SELECT
+public.ocirecipe                        = SELECT, UPDATE
+public.ocirecipearch                    = SELECT
+public.ocirecipebuild                   = SELECT, INSERT, UPDATE
+public.ocirecipebuildjob                = SELECT, UPDATE
+public.ocifile                          = SELECT
+public.ociproject                       = SELECT
+public.ociprojectname                   = SELECT
+public.ociprojectseries                 = SELECT
+public.ocipushrule                      = SELECT
+public.ociregistrycredentials           = SELECT
+public.sourcepackagename                = SELECT
+public.teammembership                   = SELECT
+public.teamparticipation                = SELECT
+public.webhook                          = SELECT
+public.webhookjob                       = SELECT, INSERT
diff --git a/lib/lp/oci/configure.zcml b/lib/lp/oci/configure.zcml
index 565412d..5836861 100644
--- a/lib/lp/oci/configure.zcml
+++ b/lib/lp/oci/configure.zcml
@@ -60,7 +60,7 @@
     <subscriber
         for="lp.oci.interfaces.ocirecipebuild.IOCIRecipeBuild
              lazr.lifecycle.interfaces.IObjectModifiedEvent"
-        handler="lp.oci.subscribers.ocirecipebuild.oci_recipe_build_status_changed" />
+        handler="lp.oci.subscribers.ocirecipebuild.oci_recipe_build_modified" />
 
     <!-- OCIRecipeBuildSet -->
     <securedutility
@@ -127,4 +127,22 @@
             interface="lp.oci.interfaces.ocipushrule.IOCIPushRuleSet"/>
     </securedutility>
 
+    <!-- OCI related jobs -->
+    <securedutility
+        component="lp.oci.model.ocirecipebuildjob.OCIRegistryUploadJob"
+        provides="lp.oci.interfaces.ocirecipebuildjob.IOCIRegistryUploadJobSource">
+        <allow interface="lp.oci.interfaces.ocirecipebuildjob.IOCIRegistryUploadJobSource" />
+    </securedutility>
+    <class class="lp.oci.model.ocirecipebuildjob.OCIRegistryUploadJob">
+        <allow interface="lp.oci.interfaces.ocirecipebuildjob.IOCIRecipeBuildJob" />
+        <allow interface="lp.oci.interfaces.ocirecipebuildjob.IOCIRegistryUploadJob" />
+    </class>
+
+    <!-- Registry interaction -->
+    <securedutility
+        class="lp.oci.model.ociregistryclient.OCIRegistryClient"
+        provides="lp.oci.interfaces.ociregistryclient.IOCIRegistryClient">
+        <allow interface="lp.oci.interfaces.ociregistryclient.IOCIRegistryClient" />
+    </securedutility>
+
 </configure>
diff --git a/lib/lp/oci/interfaces/ocirecipe.py b/lib/lp/oci/interfaces/ocirecipe.py
index 25a8967..eda2782 100644
--- a/lib/lp/oci/interfaces/ocirecipe.py
+++ b/lib/lp/oci/interfaces/ocirecipe.py
@@ -214,6 +214,12 @@ class IOCIRecipeView(Interface):
         # Really IOCIPushRule, patched in _schema_cirular_imports.
         value_type=Reference(schema=Interface), readonly=True)
 
+    can_upload_to_registry = Bool(
+        title=_("Can upload to registry"), required=True, readonly=True,
+        description=_(
+            "Whether everything is set up to allow uploading builds of "
+            "this OCI recipe to a registry."))
+
 
 class IOCIRecipeEdit(IWebhookTarget):
     """`IOCIRecipe` methods that require launchpad.Edit permission."""
diff --git a/lib/lp/oci/interfaces/ocirecipebuild.py b/lib/lp/oci/interfaces/ocirecipebuild.py
index 90bb829..a1fbfc0 100644
--- a/lib/lp/oci/interfaces/ocirecipebuild.py
+++ b/lib/lp/oci/interfaces/ocirecipebuild.py
@@ -10,12 +10,24 @@ __all__ = [
     'IOCIFile',
     'IOCIRecipeBuild',
     'IOCIRecipeBuildSet',
+    'OCIRecipeBuildRegistryUploadStatus',
     ]
 
-from lazr.restful.fields import Reference
-from zope.interface import Interface
+from lazr.enum import (
+    EnumeratedType,
+    Item,
+    )
+from lazr.restful.fields import (
+    CollectionField,
+    Reference,
+    )
+from zope.interface import (
+    Attribute,
+    Interface,
+    )
 from zope.schema import (
     Bool,
+    Choice,
     Datetime,
     Int,
     TextLine,
@@ -31,6 +43,38 @@ from lp.services.librarian.interfaces import ILibraryFileAlias
 from lp.soyuz.interfaces.distroarchseries import IDistroArchSeries
 
 
+class OCIRecipeBuildRegistryUploadStatus(EnumeratedType):
+    """OCI build registry upload status type
+
+    OCI builds may be uploaded to a registry. This represents the state of
+    that process.
+    """
+
+    UNSCHEDULED = Item("""
+        Unscheduled
+
+        No upload of this OCI build to a registry is scheduled.
+        """)
+
+    PENDING = Item("""
+        Pending
+
+        This OCI build is queued for upload to a registry.
+        """)
+
+    FAILEDTOUPLOAD = Item("""
+        Failed to upload
+
+        The last attempt to upload this OCI build to a registry failed.
+        """)
+
+    UPLOADED = Item("""
+        Uploaded
+
+        This OCI build was successfully uploaded to a registry.
+        """)
+
+
 class IOCIRecipeBuildView(IPackageBuild):
     """`IOCIRecipeBuild` attributes that require launchpad.View permission."""
 
@@ -102,6 +146,26 @@ class IOCIRecipeBuildView(IPackageBuild):
         required=True, readonly=True,
         description=_("Whether this build record can be cancelled."))
 
+    manifest = Attribute(_("The manifest of the image."))
+
+    digests = Attribute(_("File containing the image digests."))
+
+    registry_upload_jobs = CollectionField(
+        title=_("Registry upload jobs for this build."),
+        # Really IOCIRegistryUploadJob.
+        value_type=Reference(schema=Interface),
+        readonly=True)
+
+    # Really IOCIRegistryUploadJob
+    last_registry_upload_job = Reference(
+        title=_("Last registry upload job for this build."), schema=Interface)
+
+    registry_upload_status = Choice(
+        title=_("Registry upload status"),
+        vocabulary=OCIRecipeBuildRegistryUploadStatus,
+        required=True, readonly=False
+    )
+
 
 class IOCIRecipeBuildEdit(Interface):
     """`IOCIRecipeBuild` attributes that require launchpad.Edit permission."""
diff --git a/lib/lp/oci/interfaces/ocirecipebuildjob.py b/lib/lp/oci/interfaces/ocirecipebuildjob.py
index 6c25b73..7f2537b 100644
--- a/lib/lp/oci/interfaces/ocirecipebuildjob.py
+++ b/lib/lp/oci/interfaces/ocirecipebuildjob.py
@@ -8,17 +8,25 @@ from __future__ import absolute_import, print_function, unicode_literals
 __metaclass__ = type
 __all__ = [
     'IOCIRecipeBuildJob',
+    'IOCIRegistryUploadJob',
+    'IOCIRegistryUploadJobSource',
     ]
 
 from lazr.restful.fields import Reference
+from zope.component.interfaces import IObjectEvent
 from zope.interface import (
     Attribute,
     Interface,
     )
+from zope.schema import TextLine
 
 from lp import _
 from lp.oci.interfaces.ocirecipebuild import IOCIRecipeBuild
-from lp.services.job.interfaces.job import IJob
+from lp.services.job.interfaces.job import (
+    IJob,
+    IJobSource,
+    IRunnableJob,
+    )
 
 
 class IOCIRecipeBuildJob(Interface):
@@ -32,3 +40,19 @@ class IOCIRecipeBuildJob(Interface):
         schema=IOCIRecipeBuild, required=True, readonly=True)
 
     json_data = Attribute(_("A dict of data about the job."))
+
+
+class IOCIRegistryUploadJob(IRunnableJob):
+    """A Job that uploads an OCI image to a registry."""
+
+    error_message = TextLine(
+        title=_("Error message"), required=False, readonly=True)
+
+
+class IOCIRegistryUploadJobSource(IJobSource):
+
+    def create(build):
+        """Upload an OCI image to a registry.
+
+        :param build: The OCI recipe build to upload.
+        """
diff --git a/lib/lp/oci/interfaces/ociregistryclient.py b/lib/lp/oci/interfaces/ociregistryclient.py
 new file mode 100644
index 0000000..8ac2aa6
--- /dev/null
+++ b/lib/lp/oci/interfaces/ociregistryclient.py
@@ -0,0 +1,33 @@
+# Copyright 2020 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Interface for communication with an OCI registry."""
+
+from __future__ import absolute_import, print_function, unicode_literals
+
+__metaclass__ = type
+__all__ = [
+    'BlobUploadFailed',
+    'IOCIRegistryClient',
+    'ManifestUploadFailed',
+]
+
+from zope.interface import Interface
+
+
+class BlobUploadFailed(Exception):
+    pass
+
+
+class ManifestUploadFailed(Exception):
+    pass
+
+
+class IOCIRegistryClient(Interface):
+    """Interface for the API provided by an OCI registry."""
+
+    def upload(build):
+        """Upload an OCI image to a registry.
+
+        :param build: The `IOCIRecipeBuild` to upload.
+        """
diff --git a/lib/lp/oci/model/ocirecipe.py b/lib/lp/oci/model/ocirecipe.py
index 4e3273a..aa69900 100644
--- a/lib/lp/oci/model/ocirecipe.py
+++ b/lib/lp/oci/model/ocirecipe.py
@@ -377,6 +377,10 @@ class OCIRecipe(Storm, WebhookTargetMixin):
         order_by = Desc(OCIRecipeBuild.id)
         return self._getBuilds(filter_term, order_by)
 
+    @property
+    def can_upload_to_registry(self):
+        return not self.push_rules.is_empty()
+
 
 class OCIRecipeArch(Storm):
     """Link table to back `OCIRecipe.processors`."""
diff --git a/lib/lp/oci/model/ocirecipebuild.py b/lib/lp/oci/model/ocirecipebuild.py
index 595da53..c69e348 100644
--- a/lib/lp/oci/model/ocirecipebuild.py
+++ b/lib/lp/oci/model/ocirecipebuild.py
@@ -45,6 +45,11 @@ from lp.oci.interfaces.ocirecipebuild import (
     IOCIFile,
     IOCIRecipeBuild,
     IOCIRecipeBuildSet,
+    OCIRecipeBuildRegistryUploadStatus,
+    )
+from lp.oci.model.ocirecipebuildjob import (
+    OCIRecipeBuildJob,
+    OCIRecipeBuildJobType,
     )
 from lp.registry.interfaces.pocket import PackagePublishingPocket
 from lp.registry.model.person import Person
@@ -57,6 +62,8 @@ from lp.services.database.interfaces import (
     IMasterStore,
     IStore,
     )
+from lp.services.job.interfaces.job import JobStatus
+from lp.services.job.model.job import Job
 from lp.services.librarian.model import (
     LibraryFileAlias,
     LibraryFileContent,
@@ -354,23 +361,17 @@ class OCIRecipeBuild(PackageBuildMixin, Storm):
 
     @cachedproperty
     def manifest(self):
-        result = Store.of(self).find(
-            (OCIFile, LibraryFileAlias, LibraryFileContent),
-            OCIFile.build == self.id,
-            LibraryFileAlias.id == OCIFile.library_file_id,
-            LibraryFileContent.id == LibraryFileAlias.contentID,
-            LibraryFileAlias.filename == 'manifest.json')
-        return result.one()
+        try:
+            return self.getFileByName("manifest.json")
+        except NotFoundError:
+            return None
 
     @cachedproperty
     def digests(self):
-        result = Store.of(self).find(
-            (OCIFile, LibraryFileAlias, LibraryFileContent),
-            OCIFile.build == self.id,
-            LibraryFileAlias.id == OCIFile.library_file_id,
-            LibraryFileContent.id == LibraryFileAlias.contentID,
-            LibraryFileAlias.filename == 'digests.json')
-        return result.one()
+        try:
+            return self.getFileByName("digests.json")
+        except NotFoundError:
+            return None
 
     def verifySuccessfulUpload(self):
         """See `IPackageBuild`."""
@@ -383,6 +384,37 @@ class OCIRecipeBuild(PackageBuildMixin, Storm):
                             and self.digests is not None)
         return layer_files_present and metadata_present
 
+    @property
+    def registry_upload_jobs(self):
+        jobs = Store.of(self).find(
+            OCIRecipeBuildJob,
+            OCIRecipeBuildJob.build == self,
+            OCIRecipeBuildJob.job_type == OCIRecipeBuildJobType.REGISTRY_UPLOAD
+        )
+        jobs.order_by(Desc(OCIRecipeBuildJob.job_id))
+
+        def preload_jobs(rows):
+            load_related(Job, rows, ["job_id"])
+
+        return DecoratedResultSet(
+            jobs, lambda job: job.makeDerived(), pre_iter_hook=preload_jobs)
+
+    @cachedproperty
+    def last_registry_upload_job(self):
+        return self.registry_upload_jobs.first()
+
+    @property
+    def registry_upload_status(self):
+        job = self.last_registry_upload_job
+        if job is None or job.job.status == JobStatus.SUSPENDED:
+            return OCIRecipeBuildRegistryUploadStatus.UNSCHEDULED
+        elif job.job.status in (JobStatus.WAITING, JobStatus.RUNNING):
+            return OCIRecipeBuildRegistryUploadStatus.PENDING
+        elif job.job.status == JobStatus.COMPLETED:
+            return OCIRecipeBuildRegistryUploadStatus.UPLOADED
+        else:
+            return OCIRecipeBuildRegistryUploadStatus.FAILEDTOUPLOAD
+
 
 @implementer(IOCIRecipeBuildSet)
 class OCIRecipeBuildSet(SpecificBuildFarmJobSourceMixin):
diff --git a/lib/lp/oci/model/ocirecipebuildjob.py b/lib/lp/oci/model/ocirecipebuildjob.py
index 7d8699c..c17ecf3 100644
--- a/lib/lp/oci/model/ocirecipebuildjob.py
+++ b/lib/lp/oci/model/ocirecipebuildjob.py
@@ -11,20 +11,33 @@ __all__ = [
     'OCIRecipeBuildJobType',
     ]
 
+
 from lazr.delegates import delegate_to
 from lazr.enum import (
     DBEnumeratedType,
     DBItem,
     )
+from lazr.lifecycle.event import ObjectCreatedEvent
 from storm.databases.postgres import JSON
 from storm.locals import (
     Int,
     Reference,
     )
-from zope.interface import implementer
+import transaction
+from zope.component import getUtility
+from zope.event import notify
+from zope.interface import (
+    implementer,
+    provider,
+    )
 
 from lp.app.errors import NotFoundError
-from lp.oci.interfaces.ocirecipebuildjob import IOCIRecipeBuildJob
+from lp.oci.interfaces.ocirecipebuildjob import (
+    IOCIRecipeBuildJob,
+    IOCIRegistryUploadJob,
+    IOCIRegistryUploadJobSource,
+    )
+from lp.oci.interfaces.ociregistryclient import IOCIRegistryClient
 from lp.services.database.enumcol import DBEnum
 from lp.services.database.interfaces import IStore
 from lp.services.database.stormbase import StormBase
@@ -33,14 +46,13 @@ from lp.services.job.model.job import (
     Job,
     )
 from lp.services.job.runner import BaseRunnableJob
+from lp.services.propertycache import get_property_cache
+from lp.services.webapp.snapshot import notify_modified
 
 
 class OCIRecipeBuildJobType(DBEnumeratedType):
     """Values that `OCIBuildJobType.job_type` can take."""
 
-    # XXX twom (2020-04-02) This does not currently have a concrete
-    # implementation, awaiting registry upload.
-
     REGISTRY_UPLOAD = DBItem(0, """
         Registry upload
 
@@ -82,7 +94,7 @@ class OCIRecipeBuildJob(StormBase):
         self.json_data = json_data
 
     def makeDerived(self):
-        return OCIRecipeBuildJob.makeSubclass(self)
+        return OCIRecipeBuildJobDerived.makeSubclass(self)
 
 
 @delegate_to(IOCIRecipeBuildJob)
@@ -138,3 +150,101 @@ class OCIRecipeBuildJobDerived(BaseRunnableJob):
             ('oci_project_name', self.context.build.recipe.oci_project.name)
             ])
         return oops_vars
+
+
+@implementer(IOCIRegistryUploadJob)
+@provider(IOCIRegistryUploadJobSource)
+class OCIRegistryUploadJob(OCIRecipeBuildJobDerived):
+
+    class_job_type = OCIRecipeBuildJobType.REGISTRY_UPLOAD
+
+    @classmethod
+    def create(cls, build):
+        """See `IOCIRegistryUploadJobSource`"""
+        oci_build_job = OCIRecipeBuildJob(
+            build, cls.class_job_type, {})
+        job = cls(oci_build_job)
+        job.celeryRunOnCommit()
+        del get_property_cache(build).last_registry_upload_job
+        notify(ObjectCreatedEvent(build))
+        return job
+
+    # Ideally we'd just override Job._set_status or similar, but
+    # lazr.delegates makes that difficult, so we use this to override all
+    # the individual Job lifecycle methods instead.
+    def _do_lifecycle(self, method_name, manage_transaction=False,
+                      *args, **kwargs):
+        edited_fields = set()
+        with notify_modified(self.build, edited_fields) as before_modification:
+            getattr(super(OCIRegistryUploadJob, self), method_name)(
+                *args, manage_transaction=manage_transaction, **kwargs)
+            upload_status = self.build.registry_upload_status
+            if upload_status != before_modification.registry_upload_status:
+                edited_fields.add('registry_upload_status')
+        if edited_fields and manage_transaction:
+            transaction.commit()
+
+    def start(self, *args, **kwargs):
+        self._do_lifecycle("start", *args, **kwargs)
+
+    def complete(self, *args, **kwargs):
+        self._do_lifecycle("complete", *args, **kwargs)
+
+    def fail(self, *args, **kwargs):
+        self._do_lifecycle("fail", *args, **kwargs)
+
+    def queue(self, *args, **kwargs):
+        self._do_lifecycle("queue", *args, **kwargs)
+
+    def suspend(self, *args, **kwargs):
+        self._do_lifecycle("suspend", *args, **kwargs)
+
+    def resume(self, *args, **kwargs):
+        self._do_lifecycle("resume", *args, **kwargs)
+
+    @property
+    def error_message(self):
+        """See `IOCIRegistryUploadJob`."""
+        return self.json_data.get("error_message")
+
+    @error_message.setter
+    def error_message(self, message):
+        """See `IOCIRegistryUploadJob`."""
+        self.json_data["error_message"] = message
+
+    @property
+    def error_detail(self):
+        """See `IOCIRegistryUploadJob`."""
+        return self.json_data.get("error_detail")
+
+    @error_detail.setter
+    def error_detail(self, detail):
+        """See `IOCIRegistryUploadJob`."""
+        self.json_data["error_detail"] = detail
+
+    @property
+    def error_messages(self):
+        """See `IOCIRegistryUploadJob`."""
+        return self.json_data.get("error_messages")
+
+    @error_messages.setter
+    def error_messages(self, messages):
+        """See `IOCIRegistryUploadJob`."""
+        self.json_data["error_messages"] = messages
+
+    def run(self):
+        """See `IRunnableJob`."""
+        client = getUtility(IOCIRegistryClient)
+        # XXX twom 2020-04-16 This is taken from SnapStoreUploadJob
+        # it will need to gain retry support.
+        try:
+            try:
+                client.upload(self.build)
+            except Exception as e:
+                self.error_message = str(e)
+                self.error_messages = getattr(e, "messages", None)
+                self.error_detail = getattr(e, "detail", None)
+                raise
+        except Exception:
+            transaction.commit()
+            raise
diff --git a/lib/lp/oci/model/ociregistryclient.py b/lib/lp/oci/model/ociregistryclient.py
 new file mode 100644
index 0000000..2152ab8
--- /dev/null
+++ b/lib/lp/oci/model/ociregistryclient.py
@@ -0,0 +1,252 @@
+# Copyright 2020 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Client for talking to an OCI registry."""
+
+from __future__ import absolute_import, print_function, unicode_literals
+
+__metaclass__ = type
+__all__ = [
+    'OCIRegistryClient'
+]
+
+
+import hashlib
+from io import BytesIO
+import json
+import logging
+import tarfile
+
+from requests.exceptions import HTTPError
+from zope.interface import implementer
+
+from lp.oci.interfaces.ociregistryclient import (
+    BlobUploadFailed,
+    IOCIRegistryClient,
+    ManifestUploadFailed,
+    )
+from lp.services.timeout import urlfetch
+
+
+log = logging.getLogger(__name__)
+
+
+@implementer(IOCIRegistryClient)
+class OCIRegistryClient:
+
+    @classmethod
+    def _getJSONfile(cls, reference):
+        """Read JSON out of a `LibraryFileAlias`."""
+        try:
+            reference.open()
+            return json.loads(reference.read())
+        finally:
+            reference.close()
+
+    @classmethod
+    def _upload(cls, digest, push_rule, name, fileobj):
+        """Upload a blob to the registry, using a given digest.
+
+        :param digest: The digest to store the file under.
+        :param push_rule: `OCIPushRule` to use for the URL and credentials.
+        :param name: Name of the image the blob is part of.
+        :param fileobj: An object that looks like a buffer.
+
+        :raises BlobUploadFailed: if the registry does not accept the blob.
+        """
+        # Check if it already exists
+        try:
+            head_response = urlfetch(
+                "{}/v2/{}/blobs/{}".format(
+                    push_rule.registry_credentials.url, name, digest),
+                method="HEAD")
+            if head_response.status_code == 200:
+                log.info("{} already found".format(digest))
+                return
+        except HTTPError as http_error:
+            # A 404 is fine, we're about to upload the layer anyway
+            if http_error.response.status_code != 404:
+                raise http_error
+
+        post_response = urlfetch(
+            "{}/v2/{}/blobs/uploads/".format(
+                push_rule.registry_credentials.url, name),
+            method="POST")
+
+        post_location = post_response.headers["Location"]
+        query_parsed = {"digest": digest}
+
+        put_response = urlfetch(
+            post_location,
+            params=query_parsed,
+            data=fileobj,
+            method="PUT")
+
+        if put_response.status_code != 201:
+            raise BlobUploadFailed(
+                "Upload of {} for {} failed".format(digest, name))
+
+    @classmethod
+    def _upload_layer(cls, digest, push_rule, name, lfa):
+        """Upload a layer blob to the registry.
+
+        Uses _upload, but opens the LFA and extracts the necessary files
+        from the .tar.gz first.
+
+        :param digest: The digest to store the file under.
+        :param push_rule: `OCIPushRule` to use for the URL and credentials.
+        :param name: Name of the image the blob is part of.
+        :param lfa: The `LibraryFileAlias` for the layer.
+        """
+        lfa.open()
+        try:
+            un_zipped = tarfile.open(fileobj=lfa, mode='r|gz')
+            for tarinfo in un_zipped:
+                if tarinfo.name != 'layer.tar':
+                    continue
+                fileobj = un_zipped.extractfile(tarinfo)
+                cls._upload(digest, push_rule, name, fileobj)
+        finally:
+            lfa.close()
+
+    @classmethod
+    def _build_registry_manifest(cls, digests, config, config_json,
+                                 config_sha, preloaded_data):
+        """Create an image manifest for the uploading image.
+
+        This involves nearly everything as digests and lengths are required.
+        This method creates a minimal manifest, some fields are missing.
+
+        :param digests: Dict of the various digests involved.
+        :param config: The contents of the manifest config file as a dict.
+        :param config_json: The config file as a JSON string.
+        :param config_sha: The sha256sum of the config JSON string.
+        """
+        # Create the initial manifest data with empty layer information
+        manifest = {
+            "schemaVersion": 2,
+            "mediaType":
+                "application/vnd.docker.distribution.manifest.v2+json",
+            "config": {
+                "mediaType": "application/vnd.docker.container.image.v1+json",
+                "size": len(config_json),
+                "digest": "sha256:{}".format(config_sha),
+            },
+            "layers": []}
+
+        # Fill in the layer information
+        for layer in config["rootfs"]["diff_ids"]:
+            manifest["layers"].append({
+                "mediaType":
+                    "application/vnd.docker.image.rootfs.diff.tar.gzip",
+                "size": preloaded_data[layer].content.filesize,
+                "digest": layer})
+        return manifest
+
+    @classmethod
+    def _preloadFiles(cls, build, manifest, digests):
+        """Preload the data from the librarian to avoid multiple fetches
+        if there is more than one push rule for a build.
+
+        :param build: The referencing `OCIRecipeBuild`.
+        :param manifest: The manifest from the built image.
+        :param digests: Dict of the various digests involved.
+        """
+        data = {}
+        for section in manifest:
+            # Load the matching config file for this section
+            config = cls._getJSONfile(
+                build.getFileByName(section['Config']))
+            files = {"config_file": config}
+            for diff_id in config["rootfs"]["diff_ids"]:
+                # We may have already seen this diff ID.
+                if files.get(diff_id):
+                    continue
+                # Retrieve the layer files.
+                # This doesn't read the content, so there is potential
+                # for multiple fetches, but the files can be arbitrary size
+                # Potentially gigabytes.
+                files[diff_id] = {}
+                source_digest = digests[diff_id]["digest"]
+                _, lfa, _ = build.getLayerFileByDigest(source_digest)
+                files[diff_id] = lfa
+            data[section["Config"]] = files
+        return data
+
+    @classmethod
+    def _calculateTag(cls, build, push_rule):
+        """Work out the base tag for the image should be.
+
+        :param build: `OCIRecipeBuild` representing this build.
+        :param push_rule: `OCIPushRule` that we are using.
+        """
+        # XXX twom 2020-04-17 This needs to include OCIProjectSeries and
+        # base image name
+
+        return "{}".format("edge")
+
+    @classmethod
+    def upload(cls, build):
+        """Upload the artifacts from an OCIRecipeBuild to a registry.
+
+        :param build: `OCIRecipeBuild` representing this build.
+        :raises ManifestUploadFailed: If the final registry manifest fails to
+                                      upload due to network or validity.
+        """
+        # Get the required metadata files
+        manifest = cls._getJSONfile(build.manifest)
+        digests_list = cls._getJSONfile(build.digests)
+        digests = {}
+        for digest_dict in digests_list:
+            digests.update(digest_dict)
+
+        # Preload the requested files
+        preloaded_data = cls._preloadFiles(build, manifest, digests)
+
+        for push_rule in build.recipe.push_rules:
+            for section in manifest:
+                # Work out names and tags
+                image_name = push_rule.image_name
+                tag = cls._calculateTag(build, push_rule)
+                file_data = preloaded_data[section["Config"]]
+                config = file_data["config_file"]
+                #  Upload the layers involved
+                for diff_id in config["rootfs"]["diff_ids"]:
+                    cls._upload_layer(
+                        diff_id,
+                        push_rule,
+                        image_name,
+                        file_data[diff_id])
+                # The config file is required in different forms, so we can
+                # calculate the sha, work these out and upload
+                config_json = json.dumps(config).encode("UTF-8")
+                config_sha = hashlib.sha256(config_json).hexdigest()
+                cls._upload(
+                    "sha256:{}".format(config_sha),
+                    push_rule,
+                    image_name,
+                    BytesIO(config_json))
+
+                # Build the registry manifest from the image manifest
+                # and associated configs
+                registry_manifest = cls._build_registry_manifest(
+                    digests, config, config_json, config_sha,
+                    preloaded_data[section["Config"]])
+
+                # Upload the registry manifest
+                manifest_response = urlfetch(
+                    "{}/v2/{}/manifests/{}".format(
+                        push_rule.registry_credentials.url,
+                        image_name,
+                        tag),
+                    json=registry_manifest,
+                    headers={
+                        "Content-Type":
+                            "application/"
+                            "vnd.docker.distribution.manifest.v2+json"
+                        },
+                    method="PUT")
+                if manifest_response.status_code != 201:
+                    raise ManifestUploadFailed(
+                        "Failed to upload manifest for {} in {}".format(
+                            build.recipe.name, build.id))
diff --git a/lib/lp/oci/model/ociregistrycredentials.py b/lib/lp/oci/model/ociregistrycredentials.py
index 2a70159..03b0f72 100644
--- a/lib/lp/oci/model/ociregistrycredentials.py
+++ b/lib/lp/oci/model/ociregistrycredentials.py
@@ -80,8 +80,8 @@ class OCIRegistryCredentials(Storm):
     def getCredentials(self):
         container = getUtility(IEncryptedContainer, "oci-registry-secrets")
         try:
-            return json.loads(container.decrypt((
-                self._credentials['credentials_encrypted'])).decode("UTF-8"))
+            return json.loads(container.decrypt(
+                self._credentials['credentials_encrypted']).decode("UTF-8"))
         except CryptoError as e:
             # XXX twom 2020-03-18 This needs a better error
             # see SnapStoreClient.UnauthorizedUploadResponse
diff --git a/lib/lp/oci/subscribers/ocirecipebuild.py b/lib/lp/oci/subscribers/ocirecipebuild.py
index 3e7f80d..2f3fc01 100644
--- a/lib/lp/oci/subscribers/ocirecipebuild.py
+++ b/lib/lp/oci/subscribers/ocirecipebuild.py
@@ -9,8 +9,10 @@ __metaclass__ = type
 
 from zope.component import getUtility
 
+from lp.buildmaster.enums import BuildStatus
 from lp.oci.interfaces.ocirecipe import OCI_RECIPE_WEBHOOKS_FEATURE_FLAG
 from lp.oci.interfaces.ocirecipebuild import IOCIRecipeBuild
+from lp.oci.interfaces.ocirecipebuildjob import IOCIRegistryUploadJobSource
 from lp.services.features import getFeatureFlag
 from lp.services.webapp.publisher import canonical_url
 from lp.services.webhooks.interfaces import IWebhookSet
@@ -25,7 +27,7 @@ def _trigger_oci_recipe_build_webhook(build, action):
             }
         payload.update(compose_webhook_payload(
             IOCIRecipeBuild, build,
-            ["recipe", "status"]))
+            ["recipe", "status", "registry_upload_status"]))
         getUtility(IWebhookSet).trigger(
             build.recipe, "oci-recipe:build:0.1", payload)
 
@@ -34,9 +36,14 @@ def oci_recipe_build_created(build, event):
     """Trigger events when a new OCI recipe build is created."""
     _trigger_oci_recipe_build_webhook(build, "created")
 
-
-def oci_recipe_build_status_changed(build, event):
+def oci_recipe_build_modified(build, event):
     """Trigger events when OCI recipe build statuses change."""
     if event.edited_fields is not None:
-        if "status" in event.edited_fields:
+        status_changed = "status" in event.edited_fields
+        registry_changed = "registry_upload_status" in event.edited_fields
+        if status_changed or registry_changed:
             _trigger_oci_recipe_build_webhook(build, "status-changed")
+        if status_changed:
+            if (build.recipe.can_upload_to_registry and
+                    build.status == BuildStatus.FULLYBUILT):
+                getUtility(IOCIRegistryUploadJobSource).create(build)
diff --git a/lib/lp/oci/tests/test_ocirecipe.py b/lib/lp/oci/tests/test_ocirecipe.py
index e158e2e..06ed382 100644
--- a/lib/lp/oci/tests/test_ocirecipe.py
+++ b/lib/lp/oci/tests/test_ocirecipe.py
@@ -127,6 +127,7 @@ class TestOCIRecipe(TestCaseWithFactory):
             "action": Equals("created"),
             "recipe": Equals(canonical_url(recipe, force_local_path=True)),
             "status": Equals("Needs building"),
+            'registry_upload_status': Equals('Unscheduled'),
             }
         with person_logged_in(recipe.owner):
             delivery = hook.deliveries.one()
diff --git a/lib/lp/oci/tests/test_ocirecipebuild.py b/lib/lp/oci/tests/test_ocirecipebuild.py
index 400d78d..0579a8b 100644
--- a/lib/lp/oci/tests/test_ocirecipebuild.py
+++ b/lib/lp/oci/tests/test_ocirecipebuild.py
@@ -182,6 +182,7 @@ class TestOCIRecipeBuild(TestCaseWithFactory):
             "recipe": Equals(
                  canonical_url(self.build.recipe, force_local_path=True)),
             "status": Equals("Successfully built"),
+            'registry_upload_status': Equals("Unscheduled"),
             }
         self.assertThat(
             logger.output, LogsScheduledWebhooks([
diff --git a/lib/lp/oci/tests/test_ocirecipebuildjob.py b/lib/lp/oci/tests/test_ocirecipebuildjob.py
index cf85684..f4628e0 100644
--- a/lib/lp/oci/tests/test_ocirecipebuildjob.py
+++ b/lib/lp/oci/tests/test_ocirecipebuildjob.py
@@ -7,16 +7,65 @@ from __future__ import absolute_import, print_function, unicode_literals
 
 __metaclass__ = type
 
-from lp.oci.interfaces.ocirecipe import OCI_RECIPE_ALLOW_CREATE
-from lp.oci.interfaces.ocirecipebuildjob import IOCIRecipeBuildJob
+from fixtures import FakeLogger
+from testtools.matchers import (
+    Equals,
+    MatchesDict,
+    MatchesListwise,
+    MatchesStructure,
+    )
+import transaction
+from zope.interface import implementer
+
+from lp.buildmaster.enums import BuildStatus
+from lp.oci.interfaces.ocirecipe import (
+    OCI_RECIPE_ALLOW_CREATE,
+    OCI_RECIPE_WEBHOOKS_FEATURE_FLAG,
+    )
+from lp.oci.interfaces.ocirecipebuildjob import (
+    IOCIRecipeBuildJob,
+    IOCIRegistryUploadJob,
+    IOCIRegistryUploadJobSource,
+    )
+from lp.oci.interfaces.ociregistryclient import IOCIRegistryClient
 from lp.oci.model.ocirecipebuildjob import (
     OCIRecipeBuildJob,
     OCIRecipeBuildJobDerived,
     OCIRecipeBuildJobType,
+    OCIRegistryUploadJob,
     )
+from lp.services.config import config
 from lp.services.features.testing import FeatureFixture
+from lp.services.job.runner import JobRunner
 from lp.testing import TestCaseWithFactory
-from lp.testing.layers import DatabaseFunctionalLayer
+from lp.testing.dbuser import dbuser
+from lp.testing.fakemethod import FakeMethod
+from lp.testing.fixture import ZopeUtilityFixture
+from lp.testing.layers import (
+    DatabaseFunctionalLayer,
+    LaunchpadZopelessLayer,
+    )
+from lp.testing.mail_helpers import pop_notifications
+from lp.services.webapp import canonical_url
+from lp.services.webhooks.testing import LogsScheduledWebhooks
+
+
+def run_isolated_jobs(jobs):
+    """Run a sequence of jobs, ensuring transaction isolation.
+
+    We abort the transaction after each job to make sure that there is no
+    relevant uncommitted work.
+    """
+    for job in jobs:
+        JobRunner([job]).runAll()
+        transaction.abort()
+
+
+@implementer(IOCIRegistryClient)
+class FakeRegistryClient:
+
+    def __init__(self):
+        self.upload = FakeMethod()
 
 
 class FakeOCIBuildJob(OCIRecipeBuildJobDerived):
@@ -52,3 +101,104 @@ class TestOCIRecipeBuildJob(TestCaseWithFactory):
             ('oci_project_name', oci_build.recipe.oci_project.name),
             ]
         self.assertEqual(expected, oops)
+
+
+class TestOCIRegistryUploadJob(TestCaseWithFactory):
+
+    layer = LaunchpadZopelessLayer
+
+    def setUp(self):
+        super(TestOCIRegistryUploadJob, self).setUp()
+        self.useFixture(FeatureFixture({
+            'webhooks.new.enabled': 'true',
+            OCI_RECIPE_WEBHOOKS_FEATURE_FLAG: 'on',
+            OCI_RECIPE_ALLOW_CREATE: 'on'
+            }))
+
+    def makeOCIRecipeBuild(self, **kwargs):
+        ocibuild = self.factory.makeOCIRecipeBuild(
+            builder=self.factory.makeBuilder(), **kwargs)
+        ocibuild.updateStatus(BuildStatus.FULLYBUILT)
+        self.factory.makeWebhook(
+            target=ocibuild.recipe, event_types=["oci-recipe:build:0.1"])
+        return ocibuild
+
+    def assertWebhookDeliveries(self, ocibuild,
+                                expected_registry_upload_statuses, logger):
+        hook = ocibuild.recipe.webhooks.one()
+        deliveries = list(hook.deliveries)
+        deliveries.reverse()
+        expected_payloads = [{
+            "recipe_build": Equals(
+                canonical_url(ocibuild, force_local_path=True)),
+            "action": Equals("created"),
+            "recipe": Equals(
+                canonical_url(ocibuild.recipe, force_local_path=True)),
+            "status": Equals("Successfully built"),
+            "registry_upload_status": Equals("Pending")}]
+        expected_payloads += [{
+            "recipe_build": Equals(
+                canonical_url(ocibuild, force_local_path=True)),
+            "action": Equals("status-changed"),
+            "recipe": Equals(
+                canonical_url(ocibuild.recipe, force_local_path=True)),
+            "status": Equals("Successfully built"),
+            "registry_upload_status": Equals(expected),
+            } for expected in expected_registry_upload_statuses]
+        matchers = [
+            MatchesStructure(
+                event_type=Equals("oci-recipe:build:0.1"),
+                payload=MatchesDict(expected_payload))
+            for expected_payload in expected_payloads]
+        self.assertThat(deliveries, MatchesListwise(matchers))
+        with dbuser(config.IWebhookDeliveryJobSource.dbuser):
+            for delivery in deliveries:
+                self.assertEqual(
+                    "<WebhookDeliveryJob for webhook %d on %r>" % (
+                        hook.id, hook.target),
+                    repr(delivery))
+            self.assertThat(
+                logger.output, LogsScheduledWebhooks([
+                    (hook, "oci-recipe:build:0.1", MatchesDict(
+                        expected_payload))
+                    for expected_payload in expected_payloads]))
+
+    def test_provides_interface(self):
+        # `OCIRegistryUploadJob` objects provide `IOCIRegistryUploadJob`.
+        ocibuild = self.factory.makeOCIRecipeBuild()
+        job = OCIRegistryUploadJob.create(ocibuild)
+        self.assertProvides(job, IOCIRegistryUploadJob)
+
+    def test_run(self):
+        logger = self.useFixture(FakeLogger())
+        ocibuild = self.makeOCIRecipeBuild()
+        self.assertContentEqual([], ocibuild.registry_upload_jobs)
+        job = OCIRegistryUploadJob.create(ocibuild)
+        client = FakeRegistryClient()
+        self.useFixture(ZopeUtilityFixture(client, IOCIRegistryClient))
+        with dbuser(config.IOCIRegistryUploadJobSource.dbuser):
+            run_isolated_jobs([job])
+        self.assertEqual([((ocibuild,), {})], client.upload.calls)
+        self.assertContentEqual([job], ocibuild.registry_upload_jobs)
+        self.assertIsNone(job.error_message)
+        self.assertEqual([], pop_notifications())
+        self.assertWebhookDeliveries(
+            ocibuild, ["Uploaded"], logger)
+
+    def test_run_failed(self):
+        # A failed run sets the registry upload status to FAILED.
+        logger = self.useFixture(FakeLogger())
+        ocibuild = self.makeOCIRecipeBuild()
+        self.assertContentEqual([], ocibuild.registry_upload_jobs)
+        job = OCIRegistryUploadJob.create(ocibuild)
+        client = FakeRegistryClient()
+        client.upload.failure = ValueError("An upload failure")
+        self.useFixture(ZopeUtilityFixture(client, IOCIRegistryClient))
+        with dbuser(config.IOCIRegistryUploadJobSource.dbuser):
+            run_isolated_jobs([job])
+        self.assertEqual([((ocibuild,), {})], client.upload.calls)
+        self.assertContentEqual([job], ocibuild.registry_upload_jobs)
+        self.assertEqual("An upload failure", job.error_message)
+        self.assertEqual([], pop_notifications())
+        self.assertWebhookDeliveries(
+            ocibuild, ["Failed to upload"], logger)
diff --git a/lib/lp/oci/tests/test_ociregistryclient.py b/lib/lp/oci/tests/test_ociregistryclient.py
 new file mode 100644
index 0000000..5f83e38
--- /dev/null
+++ b/lib/lp/oci/tests/test_ociregistryclient.py
@@ -0,0 +1,204 @@
+# Copyright 2020 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Tests for the OCI Registry client."""
+
+from __future__ import absolute_import, print_function, unicode_literals
+
+__metaclass__ = type
+
+import json
+
+from fixtures import MockPatch
+from requests.exceptions import HTTPError
+import responses
+from testtools.matchers import (
+    Equals,
+    MatchesDict,
+    MatchesListwise,
+    )
+import transaction
+
+from lp.oci.model.ociregistryclient import OCIRegistryClient
+from lp.oci.tests.helpers import OCIConfigHelperMixin
+from lp.testing import TestCaseWithFactory
+from lp.testing.layers import LaunchpadZopelessLayer
+
+
+class TestOCIRegistryClient(OCIConfigHelperMixin, TestCaseWithFactory):
+
+    layer = LaunchpadZopelessLayer
+
+    def setUp(self):
+        super(TestOCIRegistryClient, self).setUp()
+        self.setConfig()
+        self.manifest = [{
+            "Config": "config_file_1.json",
+            "Layers": ["layer_1/layer.tar", "layer_2/layer.tar"]}]
+        self.digests = [{
+            "diff_id_1": {
+                "digest": "digest_1",
+                "source": "test/base_1",
+                "layer_id": "layer_1"
+            },
+            "diff_id_2": {
+                "digest": "digest_2",
+                "source": "",
+                "layer_id": "layer_2"
+            }
+        }]
+        self.config = {"rootfs": {"diff_ids": ["diff_id_1", "diff_id_2"]}}
+        self.build = self.factory.makeOCIRecipeBuild()
+        self.factory.makeOCIPushRule(recipe=self.build.recipe)
+        self.client = OCIRegistryClient()
+
+    def _makeFiles(self):
+        self.factory.makeOCIFile(
+            build=self.build,
+            content=json.dumps(self.manifest),
+            filename='manifest.json',
+        )
+        self.factory.makeOCIFile(
+            build=self.build,
+            content=json.dumps(self.digests),
+            filename='digests.json',
+        )
+        self.factory.makeOCIFile(
+            build=self.build,
+            content=json.dumps(self.config),
+            filename='config_file_1.json'
+        )
+
+        # make layer files
+        self.layer_1_file = self.factory.makeOCIFile(
+            build=self.build,
+            content="digest_1",
+            filename="digest_1_filename",
+            layer_file_digest="digest_1"
+        )
+        self.layer_2_file = self.factory.makeOCIFile(
+            build=self.build,
+            content="digest_2",
+            filename="digest_2_filename",
+            layer_file_digest="digest_2"
+        )
+
+        transaction.commit()
+
+    @responses.activate
+    def test_upload(self):
+        self._makeFiles()
+        self.useFixture(MockPatch(
+            "lp.oci.model.ociregistryclient.OCIRegistryClient._upload"))
+        self.useFixture(MockPatch(
+            "lp.oci.model.ociregistryclient.OCIRegistryClient._upload_layer"))
+
+        manifests_url = "{}/v2/{}/manifests/edge".format(
+            self.build.recipe.push_rules[0].registry_credentials.url,
+            self.build.recipe.push_rules[0].image_name
+        )
+        responses.add("PUT", manifests_url, status=201)
+        self.client.upload(self.build)
+
+        request = json.loads(responses.calls[0].request.body)
+
+        self.assertThat(request, MatchesDict({
+            "layers": MatchesListwise([
+                MatchesDict({
+                    "mediaType": Equals(
+                        "application/vnd.docker.image.rootfs.diff.tar.gzip"),
+                    "digest": Equals("diff_id_1"),
+                    "size": Equals(8)}),
+                MatchesDict({
+                    "mediaType": Equals(
+                        "application/vnd.docker.image.rootfs.diff.tar.gzip"),
+                    "digest": Equals("diff_id_2"),
+                    "size": Equals(8)})
+            ]),
+            "schemaVersion": Equals(2),
+            "config": MatchesDict({
+                "mediaType": Equals(
+                    "application/vnd.docker.container.image.v1+json"),
+                "digest": Equals(
+                    "sha256:33b69b4b6e106f9fc7a8b93409"
+                    "36c85cf7f84b2d017e7b55bee6ab214761f6ab"),
+                "size": Equals(52)
+            }),
+            "mediaType": Equals(
+                "application/vnd.docker.distribution.manifest.v2+json")
+        }))
+
+    def test_preloadFiles(self):
+        self._makeFiles()
+        files = self.client._preloadFiles(
+            self.build, self.manifest, self.digests[0])
+
+        self.assertThat(files, MatchesDict({
+            'config_file_1.json': MatchesDict({
+                'config_file': Equals(self.config),
+                'diff_id_1': Equals(self.layer_1_file.library_file),
+                'diff_id_2': Equals(self.layer_2_file.library_file)})}))
+
+    def test_calculateTag(self):
+        result = self.client._calculateTag(
+            self.build, self.build.recipe.push_rules[0])
+        self.assertEqual("edge", result)
+
+    def test_build_registry_manifest(self):
+        self._makeFiles()
+        preloaded_data = self.client._preloadFiles(
+            self.build, self.manifest, self.digests[0])
+        manifest = self.client._build_registry_manifest(
+            self.digests[0],
+            self.config,
+            json.dumps(self.config),
+            "config-sha",
+            preloaded_data["config_file_1.json"])
+        self.assertThat(manifest, MatchesDict({
+            "layers": MatchesListwise([
+                MatchesDict({
+                    "mediaType": Equals(
+                        "application/vnd.docker.image.rootfs.diff.tar.gzip"),
+                    "digest": Equals("diff_id_1"),
+                    "size": Equals(8)}),
+                MatchesDict({
+                    "mediaType": Equals(
+                        "application/vnd.docker.image.rootfs.diff.tar.gzip"),
+                    "digest": Equals("diff_id_2"),
+                    "size": Equals(8)})
+            ]),
+            "schemaVersion": Equals(2),
+            "config": MatchesDict({
+                "mediaType": Equals(
+                    "application/vnd.docker.container.image.v1+json"),
+                "digest": Equals("sha256:config-sha"),
+                "size": Equals(52)
+            }),
+            "mediaType": Equals(
+                "application/vnd.docker.distribution.manifest.v2+json")
+        }))
+
+    @responses.activate
+    def test_upload_handles_existing(self):
+        blobs_url = "{}/v2/{}/blobs/{}".format(
+            self.build.recipe.push_rules[0].registry_credentials.url,
+            "test-name",
+            "test-digest")
+        responses.add("HEAD", blobs_url, status=200)
+        self.client._upload(
+            "test-digest", self.build.recipe.push_rules[0], "test-name", None)
+
+    @responses.activate
+    def test_upload_raises_non_404(self):
+        blobs_url = "{}/v2/{}/blobs/{}".format(
+            self.build.recipe.push_rules[0].registry_credentials.url,
+            "test-name",
+            "test-digest")
+        responses.add("HEAD", blobs_url, status=500)
+        self.assertRaises(
+            HTTPError,
+            self.client._upload,
+            "test-digest",
+            self.build.recipe.push_rules[0],
+            "test-name",
+            None)
diff --git a/lib/lp/services/config/schema-lazr.conf b/lib/lp/services/config/schema-lazr.conf
index 18b452b..ec65b96 100644
--- a/lib/lp/services/config/schema-lazr.conf
+++ b/lib/lp/services/config/schema-lazr.conf
@@ -1821,6 +1821,7 @@ job_sources:
     IExpiringMembershipNotificationJobSource,
     IGitRepositoryModifiedMailJobSource,
     IMembershipNotificationJobSource,
+    IOCIRegistryUploadJobSource,
     IPackageUploadNotificationJobSource,
     IPersonDeactivateJobSource,
     IPersonMergeJobSource,
@@ -1984,6 +1985,11 @@ module: lp.snappy.interfaces.snapbuildjob
 dbuser: snap-build-job
 crontab_group: MAIN
 
+[IOCIRegistryUploadJobSource]
+module: lp.oci.interfaces.ocirecipebuildjob
+dbuser: oci-build-job
+crontab_group: MAIN
+
 [ITeamInvitationNotificationJobSource]
 module: lp.registry.interfaces.persontransferjob
 dbuser: person-transfer-job