Skip to content

Commit 86331f3

Browse files
committed
future: Add regression test for double callback
Add a probabilistic test that verifies the callback is invoked exactly once when set_callback races with future resolution. The test creates a future whose resolution is synchronized via a barrier, then calls set_callback at the same time, checking that the atomic callback counter never exceeds 1. The natural race window is unfortunately too narrow to hit reliably without instrumentation. Manual testing on my machine showed the bug in ~20-40% runs, so this is enough to be hit in the CI from time to time.
1 parent 797e8ec commit 86331f3

1 file changed

Lines changed: 125 additions & 0 deletions

File tree

scylla-rust-wrapper/src/future.rs

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -892,4 +892,129 @@ mod tests {
892892
let _ = unsafe { Box::from_raw(flag_ptr) };
893893
}
894894
}
895+
896+
/// Regression test for a race condition where the callback could be invoked
897+
/// twice: once by the resolution path and once by `set_callback` itself.
898+
/// The race window was between `set_callback` releasing the state lock
899+
/// (after storing the callback) and checking `result.get().is_some()`.
900+
///
901+
/// The race scenario:
902+
/// 1. The async future completes, and the resolution task tries to acquire
903+
/// the state lock.
904+
/// 2. `set_callback` acquires the lock first, stores the callback, releases
905+
/// the lock.
906+
/// 3. The resolution task immediately acquires the lock, sets result, reads
907+
/// the callback, releases the lock, and invokes the callback.
908+
/// 4. `set_callback` checks `result.get().is_some()` → true, invokes callback
909+
/// again (BUG: double invocation).
910+
///
911+
/// To trigger this, we ensure the async task has already produced its result
912+
/// (but not yet acquired the state lock to store it) by the time
913+
/// `set_callback` is called, creating lock contention.
914+
#[test]
915+
#[ntest::timeout(30000)]
916+
fn test_cass_future_callback_invoked_exactly_once() {
917+
use std::sync::Barrier;
918+
use std::sync::atomic::{AtomicU32, Ordering};
919+
920+
unsafe extern "C" fn counting_cb(
921+
_fut: CassBorrowedSharedPtr<CassFuture, CMut>,
922+
data: *mut c_void,
923+
) {
924+
let counter = unsafe { &*(data as *const AtomicU32) };
925+
counter.fetch_add(1, Ordering::SeqCst);
926+
}
927+
928+
// Use a multi-threaded runtime so the spawned future task can race
929+
// with the thread that calls set_callback.
930+
let runtime: Arc<crate::runtime::Runtime> = Arc::new(
931+
tokio::runtime::Builder::new_multi_thread()
932+
.worker_threads(4)
933+
.enable_all()
934+
.build()
935+
.unwrap()
936+
.into(),
937+
);
938+
939+
const THREADS: usize = 8;
940+
const ITERATIONS_PER_THREAD: usize = 1250;
941+
942+
// Run attempts from multiple OS threads concurrently to increase
943+
// scheduling contention and make the race more likely to manifest.
944+
let handles: Vec<_> = (0..THREADS)
945+
.map(|_| {
946+
let runtime = Arc::clone(&runtime);
947+
thread::spawn(move || {
948+
for _ in 0..ITERATIONS_PER_THREAD {
949+
let counter = Arc::new(AtomicU32::new(0));
950+
951+
let barrier = Arc::new(Barrier::new(2));
952+
let barrier_clone = Arc::clone(&barrier);
953+
954+
let fut = async move {
955+
tokio::task::spawn_blocking(move || {
956+
barrier_clone.wait();
957+
})
958+
.await
959+
.unwrap();
960+
Ok(CassResultValue::Empty)
961+
};
962+
963+
let cass_fut = CassFuture::make_raw(
964+
Arc::clone(&runtime),
965+
fut,
966+
#[cfg(cpp_integration_testing)]
967+
None,
968+
);
969+
970+
// Release the barrier (letting the async task resolve)
971+
// and immediately call set_callback — racing against
972+
// the resolution path.
973+
barrier.wait();
974+
#[allow(clippy::disallowed_methods)]
975+
let counter_ptr = Arc::as_ptr(&counter) as *mut c_void;
976+
unsafe {
977+
assert_cass_error_eq!(
978+
cass_future_set_callback(
979+
cass_fut.borrow(),
980+
Some(counting_cb),
981+
counter_ptr,
982+
),
983+
CassError::CASS_OK
984+
);
985+
}
986+
987+
// Wait for the future to be fully resolved.
988+
unsafe { cass_future_wait(cass_fut.borrow()) };
989+
990+
// Spin-wait for the callback to be invoked at least once.
991+
let deadline = std::time::Instant::now() + Duration::from_millis(100);
992+
while counter.load(Ordering::SeqCst) == 0 {
993+
assert!(
994+
std::time::Instant::now() < deadline,
995+
"Callback was never invoked"
996+
);
997+
std::thread::yield_now();
998+
}
999+
1000+
// Give a brief moment for a spurious second invocation.
1001+
std::thread::yield_now();
1002+
1003+
assert_eq!(
1004+
counter.load(Ordering::SeqCst),
1005+
1,
1006+
"Callback was invoked {} times instead of exactly once",
1007+
counter.load(Ordering::SeqCst)
1008+
);
1009+
1010+
unsafe { cass_future_free(cass_fut) };
1011+
}
1012+
})
1013+
})
1014+
.collect();
1015+
1016+
for handle in handles {
1017+
handle.join().unwrap();
1018+
}
1019+
}
8951020
}

0 commit comments

Comments
 (0)