fix(http-cache): a late single-flight leader re-reads the entry before fetching

src/cvc/ariadne/uri_http_cache.cpp

@@ -458,7 +458,7 @@ UriResult do_fetch_and_store(cvc::state &root, const std::string &key, const Uri
 // hit returns before this is called). This per-key serialization is the ONLY write coordination the
 // cache needs — the entry write itself is a single atomic node op (see the CONCURRENCY note).
 UriResult fetch_and_cache(cvc::state &root, const std::string &key, const Uri &u,
-                          const EntrySnapshot &stale, const pt::ptime &now) {
+                          const pt::ptime &now) {
   std::shared_ptr<Inflight> mine;
   {
     std::unique_lock<std::mutex> lk(flight_mutex());
@@ -474,7 +474,16 @@ UriResult fetch_and_cache(cvc::state &root, const std::string &key, const Uri &u
 
   UriResult result;
   try {
-    result = do_fetch_and_store(root, key, u, stale, now);
+    // Re-read the entry now that we lead. The caller's snapshot predates the flight lock, so a
+    // leader that stored and left between that read and our lock is invisible to it; its store
+    // happened before its erase, so this read sees it. Fresh: serve it, no second transfer.
+    // Otherwise this snapshot, not the caller's, drives the fetch: newer validators, and an entry
+    // evicted meanwhile (no-store, an invalidating store) must not make a conditional GET.
+    const EntrySnapshot snap = read_entry(root(kCacheRoot), key);
+    if (snap.present && snapshot_fresh(snap, now))
+      result = {true, snap.body, snap.effective_url, std::string()};
+    else
+      result = do_fetch_and_store(root, key, u, snap, now);
   } catch (const std::exception &e) {
     result = {false, std::string(), std::string(),
               "ari: http fetch of '" + u.raw + "' failed: " + e.what()};
@@ -520,7 +529,7 @@ UriResult cached_http_fetch(cvc::state &root, const Uri &u, const std::string &
     return {true, snap.body, snap.effective_url, std::string()}; // FRESH: no network
 
   // stale (conditional GET) or miss (unconditional) — coalesced.
-  return fetch_and_cache(root, key, u, snap, now);
+  return fetch_and_cache(root, key, u, now);
 }
 
 StoreResult invalidating_store(cvc::state &root, const Uri &u, const std::string &content,

src/cvc/tests/uri_http_cache_test.cpp

@@ -326,4 +326,38 @@ TEST_F(HttpCacheTest, SingleFlightCoalescesConcurrentResolves) {
     EXPECT_EQ(bodies[i], "C") << "follower " << i << " got the wrong body";
 }
 
+// The late arrival: a resolve that read the cache (a miss) before the leader stored, but reached
+// single-flight after the leader left, saw neither the in-flight marker nor, from its old read, the
+// entry, and became a second leader. The new leader now re-reads the entry. An instant transport
+// and a start gate make that window common; every round is one key and must be one transfer.
+TEST_F(HttpCacheTest, LateArrivalAfterTheLeaderLeftIsServedFromTheCache) {
+  auto *fake = install();
+  fake->responses.push_back(ok_resp(200, {"Cache-Control: max-age=300"}, "L")); // repeats
+  constexpr int kRounds = 1000, kThreads = 8;
+  int bad_bodies = 0;
+  for (int round = 0; round < kRounds; ++round) {
+    const std::string url = "http://ex/late/" + std::to_string(round);
+    std::atomic<int> ready{0};
+    std::atomic<bool> go{false};
+    std::vector<std::string> bodies(kThreads);
+    std::vector<std::thread> ts;
+    for (int i = 0; i < kThreads; ++i)
+      ts.emplace_back([&, i] {
+        ready.fetch_add(1);
+        while (!go.load())
+          std::this_thread::yield();
+        bodies[i] = cvc::ariadne::resolve(url).content;
+      });
+    while (ready.load() < kThreads)
+      std::this_thread::yield();
+    go.store(true);
+    for (auto &t : ts)
+      t.join();
+    for (const std::string &b : bodies)
+      bad_bodies += b == "L" ? 0 : 1;
+  }
+  EXPECT_EQ(fake->calls, kRounds) << "a resolve of an already-cached key made a second transfer";
+  EXPECT_EQ(bad_bodies, 0);
+}
+
 } // namespace