From 26e68b400d774942cedc7882171b8cab82a4cc6b Mon Sep 17 00:00:00 2001
From: Marc-Antoine Desroches <madesroches@gmail.com>
Date: Mon, 20 Jul 2026 17:50:31 -0400
Subject: [PATCH] fix: InflightManager::enqueue leaks a duplicate close flag to
 the leader

The Entry::Vacant arm created two separate Arc<AtomicBool> close flags: one
stored in the inflight table, another handed to the fetch leader. take()
(called by insert()/cancellation) only flips the table's copy, so the
leader's RawFetch::poll -- which reads its own copy -- never observes
cancellation. A racing insert() of fresh bytes can be clobbered by the
stale leader's re-insert of the value it already had in flight, since the
leader is a detached spawned task that outlives any caller scope.

Share a single Arc between the table entry and the leader instead.
---
 foyer-memory/src/inflight.rs | 39 ++++++++++++++++++++++++++++++++++--
 1 file changed, 37 insertions(+), 2 deletions(-)

diff --git a/foyer-memory/src/inflight.rs b/foyer-memory/src/inflight.rs
index 5bdd0a5d..99504c4c 100644
--- a/foyer-memory/src/inflight.rs
+++ b/foyer-memory/src/inflight.rs
@@ -207,18 +207,21 @@ where
                 let (tx, rx) = oneshot::channel();
                 let id = self.next_id;
                 self.next_id += 1;
+                // Leader and table share this Arc: `take`/`fetch_or_take` set it through the
+                // table entry, and the leader's `RawFetch::poll` reads it through its own
+                // clone, so cancellation is actually observed instead of silently no-op'd.
+                let close = Arc::new(AtomicBool::new(false));
                 let entry = InflightEntry {
                     hash,
                     key: key.to_owned(),
                     inflight: Inflight {
                         id,
-                        close: Arc::new(AtomicBool::new(false)),
+                        close: close.clone(),
                         notifiers: vec![tx],
                         f: None,
                     },
                 };
                 v.insert(entry);
-                let close = Arc::new(AtomicBool::new(false));
                 Enqueue::Lead {
                     id,
                     close,
@@ -305,3 +308,35 @@ where
     Fetch(RequiredFetchBuilder<E::Key, E::Value, E::Properties, C>),
     Notifiers(Vec<Notifier<Option<RawCacheEntry<E, S, I>>>>),
 }
+
+#[cfg(test)]
+mod tests {
+    use foyer_common::hasher::ModHasher;
+
+    use super::*;
+    use crate::{TestProperties, eviction::fifo::Fifo, indexer::hash_table::HashTableIndexer};
+
+    type TestEviction = Fifo<u64, u64, TestProperties>;
+    type TestManager = InflightManager<TestEviction, ModHasher, HashTableIndexer<TestEviction>>;
+
+    #[test]
+    fn test_enqueue_leader_observes_take_through_shared_close_flag() {
+        let mut manager = TestManager::new();
+        let key = 1u64;
+        let hash = 0;
+
+        let close = match manager.enqueue::<_, ()>(hash, &key, None) {
+            Enqueue::Lead { close, .. } => close,
+            Enqueue::Wait(_) => panic!("expected Lead on the first enqueue for a vacant key"),
+        };
+
+        // A concurrent `insert` racing the in-flight fetch cancels it via `take`, which flips
+        // the table's copy of the close flag. The leader must observe that through its own
+        // clone of the *same* Arc, or cancellation is silently inert.
+        manager.take(hash, &key, None);
+        assert!(
+            close.load(Ordering::Relaxed),
+            "leader's close flag must observe cancellation via take()"
+        );
+    }
+}