Edge caches can revalidate with If-Modified-Since after content guards

run, without downloading the file again. Last-Modified is
RepositoryContent.pulp_created for the served version, not disk mtime.
Filesystem and ArtifactResponse get Cache-Control: public, max-age=0,
must-revalidate; object-storage 302s do not. Redis can 304 from a
cached last_modified without rebuilding the body.

Signed-off-by: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com>

Author: carlosthe19916

CHANGES/7929.feature

@@ -0,0 +1 @@
+Added `Last-Modified` / `If-Modified-Since` (`304 Not Modified`) and `Cache-Control: public, max-age=0, must-revalidate` on content-app artifact responses (filesystem and `ArtifactResponse`; not object-storage 302s) so edge caches can revalidate after ContentGuard without re-fetching the body.

pulpcore/app/models/publication.py

@@ -755,14 +755,26 @@ def get_fallback_ca(self, path):
         """
         Return a ContentArtifact for path from the grace-period publication history, or None.
 
+        See :meth:`get_fallback` for the publication that contained the unit.
+        """
+        ca, _publication = self.get_fallback(path)
+        return ca
+
+    def get_fallback(self, path):
+        """
+        Return ``(ContentArtifact, Publication)`` from grace-period history, or ``(None, None)``.
+
         Iterates DistributedPublication records for this distribution from newest to oldest,
         trying each publication until the path is found.  Handles both pass-through and
         non-pass-through (PublishedArtifact) publications.
 
-        Returns None immediately when DISTRIBUTED_PUBLICATION_RETENTION_PERIOD is 0.
+        Returns ``(None, None)`` immediately when DISTRIBUTED_PUBLICATION_RETENTION_PERIOD is 0.
+        The publication is the one that still contains the unit, which may be a superseded
+        version — callers that need ``RepositoryContent.pulp_created`` must use that publication's
+        repository version, not the distribution's current one.
         """
         if not retain_distributed_pub_enabled():
-            return None
+            return None, None
         recent_dp = (
             DistributedPublication.get_non_expired()
             .filter(distribution=self)
@@ -778,7 +790,7 @@ def get_fallback_ca(self, path):
                     .first()
                 )
                 if ca is not None:
-                    return ca
+                    return ca, pub
             else:
                 pa = (
                     pub.published_artifact.select_related(
@@ -789,8 +801,8 @@ def get_fallback_ca(self, path):
                     .first()
                 )
                 if pa is not None:
-                    return pa.content_artifact
-        return None
+                    return pa.content_artifact, pub
+        return None, None
 
     @hook(AFTER_CREATE)
     @hook(

pulpcore/cache/cache.py

@@ -4,10 +4,11 @@
 from functools import wraps
 
 from aiohttp.web import FileResponse, HTTPSuccessful, Request, Response, StreamResponse
-from aiohttp.web_exceptions import HTTPFound
+from aiohttp.web_exceptions import HTTPFound, HTTPNotModified
 from django.conf import settings
 from django.http import FileResponse as ApiFileResponse
 from django.http import HttpResponse, HttpResponseRedirect
+from django.utils.http import parse_http_date_safe
 from redis import ConnectionError
 from redis.asyncio import ConnectionError as AConnectionError
 from rest_framework.request import Request as ApiRequest
@@ -18,7 +19,7 @@
     get_redis_connection,
 )
 from pulpcore.metrics import artifacts_size_counter
-from pulpcore.responses import ArtifactResponse
+from pulpcore.responses import ArtifactResponse, PulpFileResponse
 
 DEFAULT_EXPIRES_TTL = settings.CACHE_SETTINGS["EXPIRES_TTL"]
 
@@ -306,7 +307,7 @@ class AsyncContentCache(AsyncCache):
     """Cache object meant to be used for the content app"""
 
     RESPONSE_TYPES = {
-        "FileResponse": FileResponse,
+        "FileResponse": PulpFileResponse,
         "ArtifactResponse": ArtifactResponse,
         "Response": Response,
         "Redirect": HTTPFound,
@@ -349,32 +350,93 @@ async def cached_function(*args, **kwargs):
             if self.auth:
                 await self.auth(request, self, bk)
             key = self.make_key(request)
+
             # Check cache
-            response = await self.make_response(key, bk)
-            if response is None:
-                # Cache miss, create new entry
-                response = await self.make_entry(
-                    key, bk, func, args, kwargs, self.default_expires_ttl
+            entry = await self.get_entry(key, bk)
+            if entry is not None:
+                # Cache hit. Authorization has already run. If the client's If-Modified-Since
+                # covers the stored last_modified, answer a bodyless 304 without reconstructing
+                # the full response. Fall back to the header for entries cached before this field.
+                last_modified = entry.get("last_modified") or entry.get("headers", {}).get(
+                    "Last-Modified"
                 )
-            elif size := response.headers.get("X-PULP-ARTIFACT-SIZE"):
-                artifacts_size_counter.add(size)
-
+                if self._not_modified(request, last_modified):
+                    headers = dict(entry.get("headers") or {})
+                    headers["X-PULP-CACHE"] = "HIT"
+                    raise self._make_not_modified(headers, last_modified)
+                response = self.build_response(entry)
+                if size := response.headers.get("X-PULP-ARTIFACT-SIZE"):
+                    artifacts_size_counter.add(size)
+                return response
+
+            # Cache miss: build and cache the full response (a 304 is never stored). Still answer
+            # a matching conditional request with a 304 from the fresh response's Last-Modified,
+            # but never after a stream has already started writing.
+            response = await self.make_entry(key, bk, func, args, kwargs, self.default_expires_ttl)
+            if getattr(response, "prepared", False):
+                return response
+            last_modified = response.headers.get("Last-Modified")
+            if self._not_modified(request, last_modified):
+                raise self._make_not_modified(response.headers, last_modified)
             return response
 
         return cached_function
 
+    @staticmethod
+    def _not_modified(request, last_modified):
+        """True when the request's If-Modified-Since covers the given Last-Modified value.
+
+        Ignore If-Modified-Since when If-None-Match is present, or when it is later than
+        the server clock.
+        """
+        if not last_modified:
+            return False
+        if request.headers.get("If-None-Match"):
+            return False
+        if_modified_since = parse_http_date_safe(request.headers.get("If-Modified-Since", ""))
+        if if_modified_since is None or if_modified_since > time.time():
+            return False
+        lm_epoch = parse_http_date_safe(last_modified)
+        return lm_epoch is not None and lm_epoch <= if_modified_since
+
+    @staticmethod
+    def _make_not_modified(source_headers, last_modified):
+        """Build a bodyless 304 echoing Last-Modified and any caching metadata already present."""
+        headers = {"Last-Modified": last_modified}
+        for name in ("Cache-Control", "X-PULP-CACHE"):
+            if value := source_headers.get(name):
+                headers[name] = value
+        return HTTPNotModified(headers=headers)
+
     def get_request_from_args(self, args):
         """Finds the request object from list of args"""
         for arg in args:
             if isinstance(arg, Request):
                 return arg
 
-    async def make_response(self, key, base_key):
-        """Tries to find the cached entry and turn it into a proper response"""
+    async def get_entry(self, key, base_key):
+        """Return the cached entry dict for ``key`` (deleting stale/invalid rows), or None."""
         entry = await self.get(key, base_key)
         if not entry:
             return None
         entry = json.loads(entry)
+        response_type = entry.get("type")
+        # None means "doesn't expire", unset/absent means "already expired".
+        expires = entry.get("expires", -1)
+        if (not response_type or response_type not in self.RESPONSE_TYPES) or (
+            expires and expires < time.time()
+        ):
+            # Bad entry, delete from cache
+            await self.delete(key, base_key)
+            return None
+        return entry
+
+    def build_response(self, entry):
+        """Turn a cached entry dict into a proper response object (marked as a cache HIT)."""
+        entry = dict(entry)  # do not mutate the caller's dict
+        entry.pop("expires", None)
+        entry.pop("last_modified", None)
+        response_type = entry.pop("type")
 
         if binary := entry.pop("body", None):
             # raw binary data were translated to their hexadecimal representation and saved in
@@ -383,23 +445,24 @@ async def make_response(self, key, base_key):
             # https://docs.aiohttp.org/en/stable/web_reference.html#response
             entry["body"] = bytes.fromhex(binary)
 
-        response_type = entry.pop("type", None)
-        # None means "doesn't expire", unset means "already expired".
-        expires = entry.pop("expires", -1)
-        if (not response_type or response_type not in self.RESPONSE_TYPES) or (
-            expires and expires < time.time()
-        ):
-            # Bad entry, delete from cache
-            await self.delete(key, base_key)
-            return None
         response = self.RESPONSE_TYPES[response_type](**entry)
         response.headers.update({"X-PULP-CACHE": "HIT"})
         return response
 
+    async def make_response(self, key, base_key):
+        """Tries to find the cached entry and turn it into a proper response"""
+        entry = await self.get_entry(key, base_key)
+        if entry is None:
+            return None
+        return self.build_response(entry)
+
     async def make_entry(self, key, base_key, handler, args, kwargs, expires=DEFAULT_EXPIRES_TTL):
         """Gets the response for the request and try to turn it into a cacheable entry"""
         try:
             response = await handler(*args, **kwargs)
+        except HTTPNotModified:
+            # HTTPNotModified is HTTPSuccessful; do not swallow it into a cached entry.
+            raise
         except (HTTPSuccessful, HTTPFound) as e:
             response = e
 
@@ -408,7 +471,13 @@ async def make_entry(self, key, base_key, handler, args, kwargs, expires=DEFAULT
             if hasattr(response, "future_response"):
                 response = response.future_response
 
+        if getattr(response, "status", None) == 304:
+            return original_response
+
         entry = {"headers": dict(response.headers), "status": response.status}
+        if last_modified := response.headers.get("Last-Modified"):
+            # Stored alongside headers so a cache hit can 304 without reconstructing the response.
+            entry["last_modified"] = last_modified
         if expires is not None:
             # Redis TTL is not sufficient: https://github.com/pulp/pulpcore/issues/4845
             entry["expires"] = expires + time.time()