feat: let the content app return 304 Not Modified for unchanged files

pulpcore/app/util.py

@@ -17,6 +17,7 @@
 from django.conf import settings
 from django.db import connection
 from django.db.models import Model, UUIDField
+from django.utils.http import parse_http_date
 from rest_framework.reverse import reverse as drf_reverse
 from rest_framework.serializers import ValidationError
 
@@ -693,6 +694,22 @@ def normalize_http_status(status):
         return ""
 
 
+def check_request_was_modified(request, last_modified):
+    if not last_modified:
+        return True
+
+    if_modified_since = request.headers.get("If-Modified-Since")
+    if not if_modified_since:
+        return True
+
+    try:
+        last_modified_ts = parse_http_date(last_modified)
+        if_modified_ts = parse_http_date(if_modified_since)
+        return last_modified_ts > if_modified_ts
+    except (TypeError, ValueError):
+        return True
+
+
 class HashingFileWriter(RawIOBase):
     """
     A file-like object that handles writing data to disk with simultaneous

pulpcore/cache/cache.py

@@ -3,8 +3,8 @@
 import time
 from functools import wraps
 
-from aiohttp.web import FileResponse, HTTPSuccessful, Request, Response, StreamResponse
-from aiohttp.web_exceptions import HTTPFound
+from aiohttp.web import FileResponse, HTTPSuccessful, Request, Response
+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
@@ -17,6 +17,7 @@
     get_async_redis_connection,
     get_redis_connection,
 )
+from pulpcore.app.util import check_request_was_modified
 from pulpcore.metrics import artifacts_size_counter
 from pulpcore.responses import ArtifactResponse
 
@@ -350,7 +351,7 @@ async def cached_function(*args, **kwargs):
                 await self.auth(request, self, bk)
             key = self.make_key(request)
             # Check cache
-            response = await self.make_response(key, bk)
+            response = await self.make_response(key, bk, request)
             if response is None:
                 # Cache miss, create new entry
                 response = await self.make_entry(
@@ -369,7 +370,7 @@ def get_request_from_args(self, args):
             if isinstance(arg, Request):
                 return arg
 
-    async def make_response(self, key, base_key):
+    async def make_response(self, key, base_key, request=None):
         """Tries to find the cached entry and turn it into a proper response"""
         entry = await self.get(key, base_key)
         if not entry:
@@ -392,21 +393,27 @@ async def make_response(self, key, base_key):
             # Bad entry, delete from cache
             await self.delete(key, base_key)
             return None
-        response = self.RESPONSE_TYPES[response_type](**entry)
+
+        headers = entry.get("headers", {})
+        if request and not check_request_was_modified(
+            request, last_modified=headers.get("Last-Modified")
+        ):
+            response = HTTPNotModified(headers={"Cache-Control": headers.get("Cache-Control")})
+        else:
+            response = self.RESPONSE_TYPES[response_type](**entry)
         response.headers.update({"X-PULP-CACHE": "HIT"})
         return response
 
     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 (HTTPSuccessful, HTTPFound) as e:
+        except (HTTPSuccessful, HTTPFound, HTTPNotModified) as e:
             response = e
 
         original_response = response
-        if isinstance(response, StreamResponse):
-            if hasattr(response, "future_response"):
-                response = response.future_response
+        if hasattr(response, "future_response"):
+            response = response.future_response
 
         entry = {"headers": dict(response.headers), "status": response.status}
         if expires is not None:

pulpcore/content/handler.py

@@ -17,10 +17,12 @@
     HTTPFound,
     HTTPMovedPermanently,
     HTTPNotFound,
+    HTTPNotModified,
     HTTPRequestRangeNotSatisfiable,
 )
 from asgiref.sync import sync_to_async
 from django.utils import timezone
+from django.utils.http import http_date
 from multidict import CIMultiDict
 from yarl import URL
 
@@ -56,6 +58,7 @@
 )
 from pulpcore.app.util import (  # noqa: E402
     cache_key,
+    check_request_was_modified,
     get_domain,
 )
 from pulpcore.cache import AsyncContentCache  # noqa: E402
@@ -67,6 +70,8 @@
 
 log = logging.getLogger(__name__)
 
+EDGE_CACHE_CONTROL = "public, max-age=0, must-revalidate"
+
 
 class PathNotResolved(HTTPNotFound):
     """
@@ -524,6 +529,8 @@ def response_headers(path, distribution=None):
         if content_type:
             headers["Content-Type"] = content_type
 
+        headers["Cache-Control"] = EDGE_CACHE_CONTROL
+
         # Let plugin-Distribution set headers for this path if it wants.
         if distribution:
             headers.update(distribution.content_headers_for(path))
@@ -720,6 +727,9 @@ async def _match_and_stream(self, path, request):
         content_handler_result = await sync_to_async(distro.content_handler)(original_rel_path)
         if content_handler_result is not None:
             if isinstance(content_handler_result, ContentArtifact):
+                await self._add_last_modified_header(
+                    headers, content_artifact=content_handler_result, distribution=distro
+                )
                 if content_handler_result.artifact:
                     return await self._serve_content_artifact(
                         content_handler_result, headers, request
@@ -755,6 +765,9 @@ async def _match_and_stream(self, path, request):
                     raise HTTPMovedPermanently(f"{request.path}/")
                 original_rel_path = index_path
                 headers = self.response_headers(original_rel_path, distro)
+                await self._add_last_modified_header(
+                    headers, suggested_last_modified=publication.pulp_created
+                )
             except ObjectDoesNotExist:
                 dir_list, dates, sizes = await self.list_directory(None, publication, rel_path)
                 dir_list.update(
@@ -813,6 +826,11 @@ async def _match_and_stream(self, path, request):
                 except ObjectDoesNotExist:
                     pass
                 else:
+                    await self._add_last_modified_header(
+                        headers,
+                        content_artifact=ca,
+                        repository_version=publication.repository_version,
+                    )
                     if ca.artifact:
                         return await self._serve_content_artifact(ca, headers, request)
                     else:
@@ -824,6 +842,9 @@ async def _match_and_stream(self, path, request):
         if distro.SERVE_FROM_PUBLICATION:
             ca = await sync_to_async(distro.get_fallback_ca)(original_rel_path)
             if ca is not None:
+                await self._add_last_modified_header(
+                    headers, content_artifact=ca, repository_version=repo_version
+                )
                 if ca.artifact:
                     return await self._serve_content_artifact(ca, headers, request)
                 else:
@@ -871,6 +892,9 @@ async def _match_and_stream(self, path, request):
             except ObjectDoesNotExist:
                 pass
             else:
+                await self._add_last_modified_header(
+                    headers, content_artifact=ca, repository_version=repo_version
+                )
                 if ca.artifact:
                     return await self._serve_content_artifact(ca, headers, request)
                 else:
@@ -935,6 +959,38 @@ class Error(HTTPError):
             reason = None
         raise PathNotResolved(path, reason=reason)
 
+    async def _add_last_modified_header(
+        self,
+        headers,
+        suggested_last_modified: datetime | None = None,
+        content_artifact=None,
+        repository_version=None,
+        distribution=None,
+    ):
+        """
+        Add the last-modified header to the response headers if not already present.
+        """
+
+        def _find_repo_add_time():
+            if not repository_version:
+                _, rv, _ = distribution.get_repository_publication_and_version()
+            else:
+                rv = repository_version
+            cpk = content_artifact.content_id
+
+            rc = rv._content_relationships().filter(content_id=cpk).first()
+            return rc.pulp_created if rc else rv.pulp_created
+
+        if "Last-Modified" not in headers:
+            last_modified = None
+            if suggested_last_modified is not None:
+                last_modified = suggested_last_modified
+            elif content_artifact and (repository_version or distribution):
+                last_modified = await sync_to_async(_find_repo_add_time)()
+
+            if last_modified is not None:
+                headers["Last-Modified"] = http_date(last_modified.timestamp())
+
     async def _stream_content_artifact(self, request, response, content_artifact):
         """
         Stream and optionally save a ContentArtifact by requesting it using the associated remote.
@@ -1174,9 +1230,16 @@ async def _serve_content_artifact(self, content_artifact, headers, request):
             size = artifact_file.size or "*"
             raise HTTPRequestRangeNotSatisfiable(headers={"Content-Range": f"bytes */{size}"})
 
+        response = self._build_response_from_content_artifact(content_artifact, headers, request)
+
+        if not check_request_was_modified(request, last_modified=headers.get("Last-Modified")):
+            nmod_response = HTTPNotModified(headers={"Cache-Control": EDGE_CACHE_CONTROL})
+            if settings.CACHE_ENABLED:
+                nmod_response.future_response = response
+            raise nmod_response
+
         artifacts_size_counter.add(content_length)
 
-        response = self._build_response_from_content_artifact(content_artifact, headers, request)
         if isinstance(response, HTTPFound):
             raise response
         else: