fix: extend transaction_get lookup to observe all pending cache obligations

test-fixtures/src/bin/cache.rs

@@ -64,6 +64,7 @@ fn main() {
     run_test!(test_implicit_cancel_of_pending);
     run_test!(test_explicit_cancel);
     run_test!(test_collapse_across_vary);
+    run_test!(test_collapse_from_later_vary);
 
     run_test!(test_stream_back);
     run_test!(test_stream_back_fixed);
@@ -1202,6 +1203,72 @@ fn test_collapse_across_vary() {
     assert!(!txn1.pending().unwrap());
 }
 
+fn test_collapse_from_later_vary() {
+    // Per issue #626: cache contains two stale variants under Vary: Header-A.
+    // Txn1 (foo), Txn2 (bar), Txn3 (bar) dispatch; Txn1 and Txn2 receive GoGet,
+    // Txn3 waits on Txn2. Txn1 completes with an *empty* vary rule, which can in
+    // principle fulfill Txn3. Txn2 is then abandoned. Txn3 should get Found.
+    let key = new_key();
+    let header_a = HeaderName::from_static("header-a");
+
+    let b = insert(key.clone(), Duration::ZERO)
+        .header(&header_a, "foo")
+        .vary_by([&header_a])
+        .execute()
+        .unwrap();
+    b.finish().unwrap();
+
+    let b = insert(key.clone(), Duration::ZERO)
+        .header(&header_a, "bar")
+        .vary_by([&header_a])
+        .execute()
+        .unwrap();
+    b.finish().unwrap();
+
+    let pending_txn1 = Transaction::lookup(key.clone())
+        .header(&header_a, "foo")
+        .execute_async()
+        .unwrap();
+    let pending_txn2 = Transaction::lookup(key.clone())
+        .header(&header_a, "bar")
+        .execute_async()
+        .unwrap();
+    assert!(!pending_txn1.pending().unwrap(), "txn1 should have a GoGet");
+    assert!(!pending_txn2.pending().unwrap(), "txn2 should have a GoGet");
+
+    let pending_txn3 = Transaction::lookup(key.clone())
+        .header(&header_a, "bar")
+        .execute_async()
+        .unwrap();
+    assert!(pending_txn3.pending().unwrap(), "txn3 should be waiting on txn2");
+
+    let txn1 = pending_txn1.wait().unwrap();
+    let txn2 = pending_txn2.wait().unwrap();
+    assert!(txn1.must_insert_or_update());
+    assert!(txn2.must_insert_or_update());
+
+    // Txn1 completes with an empty vary rule (no .vary_by() on the insert).
+    // This pushes VaryRule::default() to the front of the key's vary_rules and
+    // inserts under the universal variant {}, which matches any request.
+    let mut writer = txn1.insert(Duration::from_secs(120)).execute().unwrap();
+    writer.write_all(b"the-response").unwrap();
+    writer.finish().unwrap();
+
+    // Txn2 is abandoned after Txn1 completes (matches the issue's ordering).
+    txn2.cancel_insert_or_update().unwrap();
+
+    // Txn3 must resolve as Found from Txn1's empty-vary insert.
+    let txn3 = pending_txn3.wait().unwrap();
+    assert!(
+        txn3.found().is_some(),
+        "txn3 should receive the cached entry from txn1's empty-vary insert"
+    );
+    assert!(
+        !txn3.must_insert_or_update(),
+        "txn3 must not be issued a GoGet"
+    );
+}
+
 fn test_stream_back() {
     let key = new_key();