过期缓存需要明确的重新验证与陈旧响应策略

nginx_lab=$(mktemp -d)
curl --fail --location --output "$nginx_lab/nginx.tar.gz" https://nginx.org/download/nginx-1.30.4.tar.gz
tar -xzf "$nginx_lab/nginx.tar.gz" -C "$nginx_lab"
cd "$nginx_lab/nginx-1.30.4"
./configure --prefix="$nginx_lab/install" --without-http_rewrite_module --without-http_gzip_module
make -j2
export NGINX_BIN="$nginx_lab/nginx-1.30.4/objs/nginx"

import hashlib
import os
from http.client import HTTPConnection
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
import signal
import socket
import subprocess
from tempfile import TemporaryDirectory
from threading import Thread
import time

state = {"body": b"A", "fail": False}
seen = []


class Origin(BaseHTTPRequestHandler):
    def log_message(self, *_):
        pass

    def do_GET(self):
        tag = '"' + hashlib.sha256(state["body"]).hexdigest() + '"'
        conditional = self.headers.get("If-None-Match")
        if self.path == "/short" and state["fail"]:
            status, body = 500, b"failure"
        elif conditional == tag:
            status, body = 304, b""
        else:
            status, body = 200, state["body"]
        directive = "max-age=1" if self.path == "/short" else "max-age=60"
        if self.path == "/no-store" or status == 500:
            directive = "no-store"
        seen.append((self.path, status, conditional))
        self.send_response(status)
        self.send_header("Cache-Control", directive)
        if status != 500:
            self.send_header("ETag", tag)
        if status != 304:
            self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)


origin = HTTPServer(("127.0.0.1", 0), Origin)
worker = Thread(target=origin.serve_forever)
worker.start()
try:
    with TemporaryDirectory(prefix="nginx-cache-lab-") as directory:
        root = Path(directory)
        with socket.socket() as reservation:
            reservation.bind(("127.0.0.1", 0))
            port = reservation.getsockname()[1]
        config = root / "nginx.conf"
        config.write_text(f"""
pid nginx.pid;
error_log stderr error;
events {{ worker_connections 64; }}
http {{
  access_log off;
  proxy_cache_path "{root}/cache" keys_zone=lab:1m max_size=10m;
  server {{
    listen 127.0.0.1:{port};
    location / {{
      proxy_pass http://127.0.0.1:{origin.server_port};
      proxy_cache lab;
      proxy_cache_valid 200 60s;
      proxy_cache_revalidate on;
      proxy_cache_use_stale http_500;
      proxy_cache_bypass $http_x_lab_bypass;
      proxy_no_cache $http_x_lab_bypass;
      proxy_connect_timeout 2s;
      proxy_read_timeout 2s;
      add_header X-Lab-Cache $upstream_cache_status always;
    }}
  }}
}}
""", encoding="utf-8")
        command = [os.environ["NGINX_BIN"], "-p", str(root) + "/",
                   "-c", str(config), "-e", "stderr"]
        checked = subprocess.run(command + ["-t"], capture_output=True, text=True)
        assert checked.returncode == 0, checked.stderr
        process = subprocess.Popen(command + ["-g", "daemon off; master_process off;"],
                                   stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True)
        try:
            deadline = time.monotonic() + 5
            while True:
                assert process.poll() is None, "nginx exited during startup"
                try:
                    with socket.create_connection(("127.0.0.1", port), timeout=0.1):
                        break
                except OSError:
                    if time.monotonic() >= deadline:
                        raise TimeoutError("nginx did not become ready")
                    time.sleep(0.05)

            def get(path, expected, body, bypass=False):
                connection = HTTPConnection("127.0.0.1", port, timeout=3)
                try:
                    connection.request("GET", path, headers={"X-Lab-Bypass": "1"} if bypass else {})
                    response = connection.getresponse()
                    actual = (response.status, response.getheader("X-Lab-Cache"), response.read())
                    assert actual == (200, expected, body), actual
                    print(path, expected, body.decode())
                finally:
                    connection.close()

            get("/long", "MISS", b"A")
            get("/long", "HIT", b"A")
            state["body"] = b"B"
            get("/long", "BYPASS", b"B", bypass=True)
            get("/long", "HIT", b"A")
            assert sum(path == "/long" for path, _, _ in seen) == 2

            get("/short", "MISS", b"B")
            time.sleep(2.1)
            get("/short", "REVALIDATED", b"B")
            assert seen[-1][1:] == (304, '"' + hashlib.sha256(b"B").hexdigest() + '"')
            state["fail"] = True
            time.sleep(2.1)
            get("/short", "STALE", b"B")
            assert seen[-1][1] == 500
            state.update(body=b"C", fail=False)
            get("/short", "EXPIRED", b"C")
            get("/short", "HIT", b"C")
            get("/no-store", "MISS", b"C")
            get("/no-store", "MISS", b"C")
            assert sum(path == "/no-store" for path, _, _ in seen) == 2
        finally:
            if process.poll() is None:
                process.send_signal(signal.SIGQUIT)
            try:
                _, errors = process.communicate(timeout=5)
            except subprocess.TimeoutExpired:
                process.kill()
                process.communicate()
                raise
            assert process.returncode == 0, errors
finally:
    origin.shutdown()
    worker.join()
    origin.server_close()