129 lines
4.3 KiB
Rust
129 lines
4.3 KiB
Rust
//! Race condition test: Immediate announcer drop
|
|
//!
|
|
//! This test verifies that all waiting listeners successfully receive
|
|
//! a value even when the announcer is dropped immediately after announce().
|
|
//!
|
|
//! Pattern being tested:
|
|
//! 1. Create announcement and listeners
|
|
//! 2. Spawn N tasks calling .listen().await
|
|
//! 3. Call announce()
|
|
//! 4. IMMEDIATELY drop announcer (before awaiting task handles)
|
|
//! 5. Verify ALL listeners successfully receive the value
|
|
//!
|
|
//! If this test FAILS, there's a race condition where the announcer dropping
|
|
//! can close the channel before all listeners read the value.
|
|
|
|
use announcement::Announcement;
|
|
use std::sync::Arc;
|
|
|
|
/// Test: Race: try_listen after drop
|
|
///
|
|
/// Verifies: Verifies try_listen() works after announcer has dropped
|
|
/// Category: race
|
|
#[tokio::test]
|
|
async fn test_immediate_drop_after_announce() {
|
|
// Test with various listener counts
|
|
for num_listeners in [1, 10, 100, 1000] {
|
|
let (announcer, announcement) = Announcement::new();
|
|
|
|
// Spawn N listeners
|
|
let mut handles = Vec::with_capacity(num_listeners);
|
|
for i in 0..num_listeners {
|
|
let listener = announcement.listener();
|
|
handles.push(tokio::spawn(async move {
|
|
listener.listen().await.map(|v: Arc<i32>| (*v, i))
|
|
}));
|
|
}
|
|
|
|
// Small delay to ensure tasks are spawned and waiting
|
|
tokio::time::sleep(tokio::time::Duration::from_millis(1)).await;
|
|
|
|
// Announce (announcer is consumed and dropped by announce())
|
|
let value = Arc::new(42);
|
|
announcer.announce(value.clone()).unwrap();
|
|
|
|
// Verify ALL listeners received the value successfully
|
|
for handle in handles {
|
|
let result = handle.await.expect("Task should not panic");
|
|
assert!(
|
|
result.is_ok(),
|
|
"Listener should successfully receive value after immediate announcer drop"
|
|
);
|
|
let (received_value, _) = result.unwrap();
|
|
assert_eq!(
|
|
received_value, 42,
|
|
"Received value should match announced value"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Test: Race: try_listen after drop
|
|
///
|
|
/// Verifies: Verifies try_listen() works after announcer has dropped
|
|
/// Category: race
|
|
#[tokio::test]
|
|
async fn test_announce_then_drop_no_delay() {
|
|
// Even more aggressive: no delay between spawn and announce
|
|
let (announcer, announcement) = Announcement::new();
|
|
|
|
let mut handles = Vec::with_capacity(100);
|
|
for i in 0..100 {
|
|
let listener = announcement.listener();
|
|
handles.push(tokio::spawn(async move {
|
|
listener.listen().await.map(|v: Arc<i32>| (*v, i))
|
|
}));
|
|
}
|
|
|
|
// NO DELAY - announce immediately (announcer consumed by announce)
|
|
let value = Arc::new(999);
|
|
announcer.announce(value.clone()).unwrap();
|
|
|
|
// All should still receive
|
|
for handle in handles {
|
|
let result = handle.await.expect("Task should not panic");
|
|
assert!(result.is_ok(), "All listeners should receive value");
|
|
assert_eq!(result.unwrap().0, 999);
|
|
}
|
|
}
|
|
|
|
/// Test: Race: try_listen after drop
|
|
///
|
|
/// Verifies: Verifies try_listen() works after announcer has dropped
|
|
/// Category: race
|
|
#[tokio::test]
|
|
async fn test_drop_before_listen_completes() {
|
|
// Test: Does listen() after announce + drop work?
|
|
let (announcer, announcement) = Announcement::new();
|
|
let listener = announcement.listener();
|
|
|
|
let value = Arc::new(777);
|
|
announcer.announce(value.clone()).unwrap();
|
|
// Announcer consumed by announce()
|
|
|
|
// This should still work - value is already available
|
|
let result = listener.listen().await;
|
|
assert!(result.is_ok(), "Should receive value even after drop");
|
|
assert_eq!(*result.unwrap(), 777);
|
|
}
|
|
|
|
/// Test: Race: try_listen after drop
|
|
///
|
|
/// Verifies: Verifies try_listen() works after announcer has dropped
|
|
/// Category: race
|
|
#[test]
|
|
fn test_try_listen_after_drop() {
|
|
// Synchronous version: try_listen after drop
|
|
let (announcer, announcement) = Announcement::new();
|
|
let listener = announcement.listener();
|
|
|
|
let value = Arc::new(555);
|
|
announcer.announce(value.clone()).unwrap();
|
|
// Announcer consumed by announce()
|
|
|
|
// try_listen should work
|
|
let result = listener.try_listen();
|
|
assert!(result.is_some(), "Should get value via try_listen");
|
|
assert_eq!(*result.unwrap(), 555);
|
|
}
|