diff --git a/data/cache/cache_key.py b/data/cache/cache_key.py
index 2a581c0926..b86bfd0f2e 100644
--- a/data/cache/cache_key.py
+++ b/data/cache/cache_key.py
@@ -67,7 +67,7 @@ def for_security_report(digest, cache_config):
 
     # Security reports don't change often so a longer TTL can be justified.
     cache_ttl = cache_config.get("security_report_cache_ttl", "300s")
-    return CacheKey(f"security_report_{digest}", cache_ttl)
+    return CacheKey(f"security_report__{digest}", cache_ttl)
 
 
 def for_repository_lookup(namespace_name, repo_name, manifest_ref, kind_filter, cache_config):
@@ -85,3 +85,11 @@ def for_repository_lookup(namespace_name, repo_name, manifest_ref, kind_filter,
 
     logger.debug(f"Loading repository lookup from cache_key: {cache_key}")
     return CacheKey(cache_key, cache_ttl)
+
+
+def for_repository_manifest(repository_id, digest, cache_config):
+    """
+    Returns a cache key for the manifest of a repository.
+    """
+    cache_ttl = cache_config.get("repository_manifest_cache_ttl", "300s")
+    return CacheKey("repository_manifest__%s_%s" % (repository_id, digest), cache_ttl)
diff --git a/data/cache/impl.py b/data/cache/impl.py
index 4c9d666dcd..cc29c4c11e 100644
--- a/data/cache/impl.py
+++ b/data/cache/impl.py
@@ -51,6 +51,10 @@ def retrieve(self, cache_key, loader, should_cache=is_not_none):
         """
         pass
 
+    @abstractmethod
+    def invalidate(self, cache_key):
+        pass
+
 
 class DisconnectWrapper(DataModelCache):
     """
@@ -67,6 +71,10 @@ def retrieve(self, cache_key, loader, should_cache=is_not_none):
         with CloseForLongOperation(self.app_config):
             return self.cache.retrieve(cache_key, loader, should_cache)
 
+    def invalidate(self, cache_key):
+        with CloseForLongOperation(self.app_config):
+            return self.cache.invalidate(cache_key)
+
 
 class NoopDataModelCache(DataModelCache):
     """
@@ -76,6 +84,9 @@ class NoopDataModelCache(DataModelCache):
     def retrieve(self, cache_key, loader, should_cache=is_not_none):
         return loader()
 
+    def invalidate(self, cache_key):
+        return
+
 
 class InMemoryDataModelCache(DataModelCache):
     """
@@ -124,6 +135,12 @@ def retrieve(self, cache_key, loader, should_cache=is_not_none):
 
         return result
 
+    def invalidate(self, cache_key):
+        try:
+            del self.cache[cache_key.key]
+        except KeyError:
+            pass
+
 
 _DEFAULT_MEMCACHE_TIMEOUT = 1  # second
 _DEFAULT_MEMCACHE_CONNECT_TIMEOUT = 1  # second
@@ -238,6 +255,14 @@ def retrieve(self, cache_key, loader, should_cache=is_not_none):
 
         return result
 
+    def invalidate(self, cache_key):
+        client = self.client_pool
+        if client is not None:
+            try:
+                client.delete(cache_key.key, True)
+            except:
+                pass
+
 
 class RedisDataModelCache(DataModelCache):
     """
@@ -315,3 +340,7 @@ def retrieve(self, cache_key, loader, should_cache=is_not_none):
             logger.debug("Not caching loaded result for key %s: %s", cache_key.key, result)
 
         return result
+
+    def invalidate(self, cache_key):
+        if self.client is not None:
+            self.client.delete(cache_key.key)
diff --git a/data/cache/redis_cache.py b/data/cache/redis_cache.py
index 052dfe159c..0da78249e5 100644
--- a/data/cache/redis_cache.py
+++ b/data/cache/redis_cache.py
@@ -32,6 +32,9 @@ def get(self, key, *args, **kwargs):
     def set(self, key, val, *args, **kwargs):
         return self.write_client.set(key, val, *args, **kwargs)
 
+    def delete(self, key, *args, **kwargs):
+        return self.write_client.delete(key, *args, **kwargs)
+
     def __getattr__(self, name):
         return getattr(self.write_client, name, None)
 
diff --git a/data/model/test/test_gc.py b/data/model/test/test_gc.py
index 72ecd3e5b1..3222723c4e 100644
--- a/data/model/test/test_gc.py
+++ b/data/model/test/test_gc.py
@@ -3,15 +3,13 @@
 import random
 from contextlib import contextmanager
 from datetime import datetime, timedelta
-from test.fixtures import *
-from test.helpers import check_transitive_modifications
 
 import pytest
 from freezegun import freeze_time
 from mock import patch
 from playhouse.test_utils import assert_query_count
 
-from app import docker_v2_signing_key, storage
+from app import docker_v2_signing_key, model_cache, storage
 from data import database, model
 from data.database import (
     ApprBlob,
@@ -32,6 +30,8 @@
 from image.oci.config import OCIConfig
 from image.oci.manifest import OCIManifestBuilder
 from image.shared.schemas import parse_manifest_from_bytes
+from test.fixtures import *
+from test.helpers import check_transitive_modifications
 from util.bytes import Bytes
 
 ADMIN_ACCESS_USER = "devtable"
@@ -84,7 +84,7 @@ def gc_now(repository):
 
 def delete_tag(repository, tag, perform_gc=True, expect_gc=True):
     repo_ref = RepositoryReference.for_repo_obj(repository)
-    registry_model.delete_tag(repo_ref, tag)
+    registry_model.delete_tag(model_cache, repo_ref, tag)
     if perform_gc:
         assert gc_now(repository) == expect_gc
 
diff --git a/data/registry_model/registry_oci_model.py b/data/registry_model/registry_oci_model.py
index 69f8aeb8a1..1e6277daa4 100644
--- a/data/registry_model/registry_oci_model.py
+++ b/data/registry_model/registry_oci_model.py
@@ -41,6 +41,7 @@
 )
 from image.docker.schema2 import EMPTY_LAYER_BLOB_DIGEST, EMPTY_LAYER_BYTES
 from image.shared import ManifestException
+from util.bytes import Bytes
 from util.timedeltastring import convert_to_timedelta
 
 logger = logging.getLogger(__name__)
@@ -163,6 +164,7 @@ def lookup_manifest_by_digest(
             allow_hidden=allow_hidden,
             require_available=require_available,
         )
+
         if manifest is None:
             if raise_on_error:
                 raise model.ManifestDoesNotExist()
@@ -528,7 +530,7 @@ def retarget_tag(
 
             return Tag.for_tag(tag, self._legacy_image_id_handler)
 
-    def delete_tag(self, repository_ref, tag_name):
+    def delete_tag(self, model_cache, repository_ref, tag_name):
         """
         Deletes the latest, *active* tag with the given name in the repository.
         """
@@ -537,9 +539,14 @@ def delete_tag(self, repository_ref, tag_name):
             if deleted_tag is None:
                 return None
 
+            manifest_cache_key = cache_key.for_repository_manifest(
+                deleted_tag.repository.id, deleted_tag.manifest.digest, model_cache.cache_config
+            )
+            model_cache.invalidate(manifest_cache_key)
+
             return Tag.for_tag(deleted_tag, self._legacy_image_id_handler)
 
-    def delete_tags_for_manifest(self, manifest):
+    def delete_tags_for_manifest(self, model_cache, manifest):
         """
         Deletes all tags pointing to the given manifest, making the manifest inaccessible for
         pulling.
@@ -547,6 +554,11 @@ def delete_tags_for_manifest(self, manifest):
         Returns the tags (ShallowTag) deleted. Returns None on error.
         """
         with db_disallow_replica_use():
+            manifest_cache_key = cache_key.for_repository_manifest(
+                manifest.repository.id, manifest.digest, model_cache.cache_config
+            )
+            model_cache.invalidate(manifest_cache_key)
+
             deleted_tags = oci.tag.delete_tags_for_manifest(manifest._db_id)
             return [ShallowTag.for_tag(tag) for tag in deleted_tags]
 
@@ -825,6 +837,51 @@ def is_namespace_enabled(self, namespace_name):
         namespace = model.user.get_namespace_user(namespace_name)
         return namespace is not None and namespace.enabled
 
+    def lookup_cached_manifest_by_digest(
+        self,
+        model_cache,
+        repository_ref,
+        manifest_digest,
+        allow_dead=False,
+        allow_hidden=False,
+        require_available=False,
+        raise_on_error=False,
+    ):
+        def load_manifest():
+            manifest = self.lookup_manifest_by_digest(
+                repository_ref,
+                manifest_digest,
+                allow_dead,
+                allow_hidden,
+                require_available,
+                raise_on_error,
+            )
+
+            if manifest:
+                manifest_dict = manifest.asdict()
+                manifest_dict["internal_manifest_bytes"] = manifest_dict[
+                    "internal_manifest_bytes"
+                ].as_unicode()
+                manifest_dict["inputs"]["repository"] = manifest_dict["inputs"][
+                    "repository"
+                ].asdict()
+                manifest_dict["inputs"]["legacy_image_handler"] = None  # TODO(kleesc): Remove
+                manifest_dict["inputs"]["legacy_id_handler"] = None  # TODO(kleesc): Remove
+
+                return manifest_dict
+
+        manifest_cache_key = cache_key.for_repository_manifest(
+            repository_ref.id, manifest_digest, model_cache.cache_config
+        )
+
+        result = model_cache.retrieve(manifest_cache_key, load_manifest)
+        # TODO(kleesc): cleanup this Manifest interface to avoid explicit conversions
+        result["internal_manifest_bytes"] = Bytes.for_string_or_unicode(
+            result["internal_manifest_bytes"]
+        )
+
+        return Manifest.from_dict(result)
+
     def lookup_cached_active_repository_tags(
         self, model_cache, repository_ref, start_pagination_id, limit
     ):
diff --git a/data/registry_model/test/test_interface.py b/data/registry_model/test/test_interface.py
index 335cdb19b8..c34f41e39b 100644
--- a/data/registry_model/test/test_interface.py
+++ b/data/registry_model/test/test_interface.py
@@ -11,7 +11,7 @@
 from mock import patch
 from playhouse.test_utils import assert_query_count
 
-from app import docker_v2_signing_key, storage
+from app import docker_v2_signing_key, model_cache, storage
 from data import model
 from data.cache import cache_key
 from data.cache.impl import InMemoryDataModelCache
@@ -368,11 +368,11 @@ def test_delete_tags(repo_namespace, repo_name, via_manifest, registry_model):
     # Delete every tag in the repository.
     for tag in tags:
         if via_manifest:
-            assert registry_model.delete_tag(repository_ref, tag.name)
+            assert registry_model.delete_tag(model_cache, repository_ref, tag.name)
         else:
             manifest = registry_model.get_manifest_for_tag(tag)
             if manifest is not None:
-                registry_model.delete_tags_for_manifest(manifest)
+                registry_model.delete_tags_for_manifest(model_cache, manifest)
 
         # Make sure the tag is no longer found.
         with assert_query_count(1):
diff --git a/endpoints/api/tag.py b/endpoints/api/tag.py
index 3243dfc856..353c03b2c2 100644
--- a/endpoints/api/tag.py
+++ b/endpoints/api/tag.py
@@ -5,7 +5,7 @@
 
 from flask import abort, request
 
-from app import app, docker_v2_signing_key, storage
+from app import app, docker_v2_signing_key, model_cache, storage
 from auth.auth_context import get_authenticated_user
 from data.model import repository as repository_model
 from data.registry_model import registry_model
@@ -248,7 +248,7 @@ def delete(self, namespace, repository, tag):
         if repo_ref is None:
             raise NotFound()
 
-        tag_ref = registry_model.delete_tag(repo_ref, tag)
+        tag_ref = registry_model.delete_tag(model_cache, repo_ref, tag)
         if tag_ref is None:
             raise NotFound()
 
diff --git a/endpoints/v1/tag.py b/endpoints/v1/tag.py
index 8ca7af3dcb..bdaccdbdb6 100644
--- a/endpoints/v1/tag.py
+++ b/endpoints/v1/tag.py
@@ -3,7 +3,7 @@
 
 from flask import abort, jsonify, make_response, request, session
 
-from app import docker_v2_signing_key, storage
+from app import docker_v2_signing_key, model_cache, storage
 from auth.decorators import process_auth
 from auth.permissions import ModifyRepositoryPermission, ReadRepositoryPermission
 from data.registry_model import registry_model
@@ -129,7 +129,7 @@ def delete_tag(namespace_name, repo_name, tag):
     )
 
     if permission.can() and repository_ref is not None:
-        if not registry_model.delete_tag(repository_ref, tag):
+        if not registry_model.delete_tag(model_cache, repository_ref, tag):
             abort(404)
 
         track_and_log("delete_tag", repository_ref, tag=tag)
diff --git a/endpoints/v2/manifest.py b/endpoints/v2/manifest.py
index e547d12f19..3ecdaba1e2 100644
--- a/endpoints/v2/manifest.py
+++ b/endpoints/v2/manifest.py
@@ -158,7 +158,8 @@ def fetch_manifest_by_digest(namespace_name, repo_name, manifest_ref, registry_m
         raise NameUnknown("repository not found")
 
     try:
-        manifest = registry_model.lookup_manifest_by_digest(
+        manifest = registry_model.lookup_cached_manifest_by_digest(
+            model_cache,
             repository_ref,
             manifest_ref,
             raise_on_error=True,
@@ -391,7 +392,7 @@ def delete_manifest_by_digest(namespace_name, repo_name, manifest_ref):
         if manifest is None:
             raise ManifestUnknown()
 
-        tags = registry_model.delete_tags_for_manifest(manifest)
+        tags = registry_model.delete_tags_for_manifest(model_cache, manifest)
         if not tags:
             raise ManifestUnknown()
 
diff --git a/endpoints/v2/test/test_manifest_pullthru.py b/endpoints/v2/test/test_manifest_pullthru.py
index bc58920b24..ab0e3860c4 100644
--- a/endpoints/v2/test/test_manifest_pullthru.py
+++ b/endpoints/v2/test/test_manifest_pullthru.py
@@ -1,12 +1,11 @@
 import unittest
-from test.fixtures import *  # noqa: F401, F403
 from unittest.mock import MagicMock, patch
 
 import pytest
 from flask import url_for
 
 from app import app as realapp
-from app import instance_keys
+from app import instance_keys, model_cache
 from auth.auth_context_type import ValidatedAuthContext
 from data import model
 from data.database import (
@@ -38,6 +37,7 @@
 )
 from image.shared.schemas import parse_manifest_from_bytes
 from proxy.fixtures import *  # noqa: F401, F403
+from test.fixtures import *  # noqa: F401, F403
 from util.bytes import Bytes
 from util.security.registry_jwt import build_context_and_subject, generate_bearer_token
 
@@ -332,6 +332,8 @@ class TestManifestPullThroughStorage:
 
     @pytest.fixture(autouse=True)
     def setup(self, client, app):
+        model_cache.empty_for_testing()
+
         self.client = client
 
         self.user = model.user.get_user("devtable")