Allow cache hit filter to run on stale meta

This changes the cache hit filter to run on both fresh and stale cache
hits so implementors can choose force invalidation behavior on stale
assets as well.

Author: drcaramelsyrup

.bleep

@@ -1 +1 @@
-2513fe1ff4219fbb218dfd1fe7ffb4bbfb96aa91
+ceca785a0ae9a5fe7d33d9b75ce003caa941b727

pingora-proxy/src/proxy_cache.rs

@@ -22,6 +22,7 @@ use pingora_core::protocols::http::conditional_filter::to_304;
 use pingora_core::protocols::http::v1::common::header_value_content_length;
 use pingora_core::ErrorType;
 use range_filter::RangeBodyFilter;
+use std::time::SystemTime;
 
 impl<SV> HttpProxy<SV> {
     // return bool: server_session can be reused, and error if any
@@ -113,29 +114,35 @@ impl<SV> HttpProxy<SV> {
 
                         // hit
                         // TODO: maybe round and/or cache now()
-                        let hit_status = if meta.is_fresh(std::time::SystemTime::now()) {
-                            // check if we should force expire or miss
-                            // vs. hard purge which forces miss)
-                            match self.inner.cache_hit_filter(session, &meta, ctx).await {
-                                Err(e) => {
-                                    error!(
-                                        "Failed to filter cache hit: {e}, {}",
-                                        self.inner.request_summary(session, ctx)
-                                    );
-                                    // this return value will cause us to fetch from upstream
-                                    HitStatus::FailedHitFilter
-                                }
-                                Ok(None) => HitStatus::Fresh,
-                                Ok(Some(ForcedInvalidationKind::ForceExpired)) => {
-                                    // force expired asset should not be serve as stale
-                                    // because force expire is usually to remove data
-                                    meta.disable_serve_stale();
-                                    HitStatus::ForceExpired
+                        let is_fresh = meta.is_fresh(SystemTime::now());
+                        // check if we should force expire or force miss
+                        let hit_status = match self
+                            .inner
+                            .cache_hit_filter(session, &meta, is_fresh, ctx)
+                            .await
+                        {
+                            Err(e) => {
+                                error!(
+                                    "Failed to filter cache hit: {e}, {}",
+                                    self.inner.request_summary(session, ctx)
+                                );
+                                // this return value will cause us to fetch from upstream
+                                HitStatus::FailedHitFilter
+                            }
+                            Ok(None) => {
+                                if is_fresh {
+                                    HitStatus::Fresh
+                                } else {
+                                    HitStatus::Expired
                                 }
-                                Ok(Some(ForcedInvalidationKind::ForceMiss)) => HitStatus::ForceMiss,
                             }
-                        } else {
-                            HitStatus::Expired
+                            Ok(Some(ForcedInvalidationKind::ForceExpired)) => {
+                                // force expired asset should not be serve as stale
+                                // because force expire is usually to remove data
+                                meta.disable_serve_stale();
+                                HitStatus::ForceExpired
+                            }
+                            Ok(Some(ForcedInvalidationKind::ForceMiss)) => HitStatus::ForceMiss,
                         };
 
                         hit_status_opt = Some(hit_status);

pingora-proxy/src/proxy_trait.rs

@@ -138,13 +138,15 @@ pub trait ProxyHttp {
     /// cache asset is ready to be used.
     ///
     /// This filter allows the user to log or force invalidate the asset.
+    /// This also runs on stale hit assets (for which `is_fresh` is false).
     ///
     /// The value returned indicates if the force invalidation should be used,
     /// and which kind. Returning `None` indicates no forced invalidation
     async fn cache_hit_filter(
         &self,
         _session: &Session,
         _meta: &CacheMeta,
+        _is_fresh: bool,
         _ctx: &mut Self::CTX,
     ) -> Result<Option<ForcedInvalidationKind>>
     where

pingora-proxy/tests/test_upstream.rs

@@ -477,6 +477,48 @@ mod test_cache {
         assert_eq!(res.text().await.unwrap(), "hello world");
     }
 
+    #[tokio::test]
+    async fn test_force_miss_stale() {
+        init();
+        let url = "http://127.0.0.1:6148/unique/test_froce_miss_stale/revalidate_now";
+
+        let res = reqwest::get(url).await.unwrap();
+        assert_eq!(res.status(), StatusCode::OK);
+        let headers = res.headers();
+        let cache_miss_epoch = headers["x-epoch"].to_str().unwrap().parse::<f64>().unwrap();
+        assert_eq!(headers["x-cache-status"], "miss");
+        assert_eq!(headers["x-upstream-status"], "200");
+        assert_eq!(res.text().await.unwrap(), "hello world");
+
+        let res = reqwest::get(url).await.unwrap();
+        assert_eq!(res.status(), StatusCode::OK);
+        let headers = res.headers();
+        let cache_hit_epoch = headers["x-epoch"].to_str().unwrap().parse::<f64>().unwrap();
+        assert_eq!(headers["x-cache-status"], "hit");
+        assert!(headers.get("x-upstream-status").is_none());
+        assert_eq!(res.text().await.unwrap(), "hello world");
+
+        assert_eq!(cache_miss_epoch, cache_hit_epoch);
+
+        sleep(Duration::from_millis(1100)).await; // ttl is 1
+
+        // stale, but can be forced miss
+        let res = reqwest::Client::new()
+            .get(url)
+            .header("x-force-miss", "1")
+            .send()
+            .await
+            .unwrap();
+
+        assert_eq!(res.status(), StatusCode::OK);
+        let headers = res.headers();
+        assert_eq!(headers["x-cache-status"], "miss");
+        assert_eq!(headers["x-upstream-status"], "200");
+        let cache_miss_epoch2 = headers["x-epoch"].to_str().unwrap().parse::<f64>().unwrap();
+        assert!(cache_miss_epoch != cache_miss_epoch2);
+        assert_eq!(res.text().await.unwrap(), "hello world");
+    }
+
     #[tokio::test]
     async fn test_cache_downstream_revalidation_etag() {
         init();

pingora-proxy/tests/utils/server_utils.rs

@@ -415,12 +415,19 @@ impl ProxyHttp for ExampleProxyCache {
         &self,
         session: &Session,
         _meta: &CacheMeta,
+        is_fresh: bool,
         _ctx: &mut Self::CTX,
     ) -> Result<Option<ForcedInvalidationKind>> {
         // allow test header to control force expiry/miss
         if session.get_header_bytes("x-force-miss") != b"" {
             return Ok(Some(ForcedInvalidationKind::ForceMiss));
         }
+
+        if !is_fresh {
+            // already expired
+            return Ok(None);
+        }
+
         if session.get_header_bytes("x-force-expire") != b"" {
             return Ok(Some(ForcedInvalidationKind::ForceExpired));
         }