fix(layers/foyer): skip cached entries for conditional reads

core/layers/foyer/src/full.rs

@@ -135,6 +135,25 @@ impl FullReader {
             return Ok(None);
         }
 
+        // A cached entry carries no precondition, so a conditional read must
+        // not be served from it: fall back to the origin, which observes the
+        // original args. On a miss the fill below still validates the
+        // condition at the origin, so only the hit path is skipped.
+        // `version` is not listed here because it is part of the cache key.
+        if self.args.if_match().is_some()
+            || self.args.if_none_match().is_some()
+            || self.args.if_version_match().is_some()
+            || self.args.if_version_not_match().is_some()
+            || self.args.if_modified_since().is_some()
+            || self.args.if_unmodified_since().is_some()
+        {
+            match self.inner.cache.get(&key).await {
+                Ok(Some(_)) => return Ok(None),
+                Ok(None) => {}
+                Err(e) => return Err(extract_err(e)),
+            }
+        }
+
         let result = self
             .inner
             .cache

core/layers/foyer/src/lib.rs

@@ -714,6 +714,45 @@ mod tests {
         assert_eq!(read_args.if_match(), Some("etag-1"));
     }
 
+    #[tokio::test]
+    async fn test_conditional_read_bypasses_full_object_cache() {
+        let cache = memory_cache().await;
+        let source = Arc::new(MockReadService::new("0123456789"));
+        let state = source.state.clone();
+        let service = FoyerLayer::new(cache)
+            .with_size_limit(0..100)
+            .apply_service(source);
+        let ctx = service_context(&service);
+
+        // Fill the cache with an unconditional read.
+        let reader = service.read(&ctx, "test", OpRead::default()).unwrap();
+        let (_, mut stream) = reader.open(BytesRange::from(0_u64..2)).await.unwrap();
+        let buffer = stream.read_all().await.unwrap();
+        assert_eq!(buffer.to_vec(), b"01");
+        assert_eq!(state.stat_calls.load(Ordering::Relaxed), 1);
+        assert_eq!(state.open_calls.load(Ordering::Relaxed), 1);
+
+        // A conditional read must not be served from the cache: the cached
+        // entry carries no condition, so only the origin can observe it.
+        let (_, args, _) = options::ReadOptions {
+            if_match: Some("etag-1".to_owned()),
+            ..Default::default()
+        }
+        .into();
+        let reader = service.read(&ctx, "test", args).unwrap();
+        let (_, mut stream) = reader.open(BytesRange::from(0_u64..2)).await.unwrap();
+        let buffer = stream.read_all().await.unwrap();
+        assert_eq!(buffer.to_vec(), b"01");
+        assert_eq!(state.open_calls.load(Ordering::Relaxed), 2);
+
+        // Unconditional reads still hit the cache.
+        let reader = service.read(&ctx, "test", OpRead::default()).unwrap();
+        let (_, mut stream) = reader.open(BytesRange::from(0_u64..2)).await.unwrap();
+        let buffer = stream.read_all().await.unwrap();
+        assert_eq!(buffer.to_vec(), b"01");
+        assert_eq!(state.open_calls.load(Ordering::Relaxed), 2);
+    }
+
     #[tokio::test]
     async fn test() {
         let dir = tempfile::tempdir().unwrap();