Skip to main content

max / makenotwork

6.3 KB · 180 lines History Blame Raw
1 //! Concurrent churn against the hub.
2 //!
3 //! The hub deliberately diverges from the SyncKit SSE pruning pattern, and the
4 //! reason is a race that only appears under contention: read `receiver_count()`,
5 //! drop the guard, then `remove()`, and a subscriber arriving between those steps
6 //! is left holding a receiver whose sender has just been dropped from the map.
7 //! Nothing errors. The subscriber simply never receives anything again.
8 //!
9 //! A single-threaded test cannot see that, so the claim is worth nothing without
10 //! these. Each test here fails against the read-then-remove implementation and
11 //! passes against `remove_if`.
12
13 use std::sync::Arc;
14 use std::sync::atomic::{AtomicUsize, Ordering};
15
16 use livechat::{ChatEvent, Hub, HubLimits, MessageId, RoomId, UserId};
17 use uuid::Uuid;
18
19 fn event(id: i64) -> ChatEvent {
20 ChatEvent::Delete { id: MessageId(id) }
21 }
22
23 /// Subscribe, immediately publish, and require delivery, while other tasks churn
24 /// the same room's channel in and out of existence.
25 ///
26 /// This is the orphaned-subscriber failure stated directly: if pruning can race a
27 /// subscribe, some iteration attaches to a sender that is about to be removed and
28 /// the publish goes to a fresh channel it is not listening to.
29 #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
30 async fn a_subscriber_is_never_orphaned_by_a_concurrent_prune() {
31 let hub = Hub::new(HubLimits {
32 max_connections_per_user: 512,
33 max_connections_total: 4096,
34 room_buffer: 64,
35 });
36 let room = RoomId(Uuid::new_v4());
37 let missed = Arc::new(AtomicUsize::new(0));
38
39 let mut tasks = Vec::new();
40 for _ in 0..16 {
41 let hub = hub.clone();
42 let missed = missed.clone();
43 tasks.push(tokio::spawn(async move {
44 for i in 0..200 {
45 let mut sub = hub.subscribe(room, UserId(Uuid::new_v4())).unwrap();
46
47 // Publish to our own subscription. Delivery is not optional: we
48 // hold a live receiver on this room.
49 let delivered = hub.publish(room, event(i));
50 if delivered == 0 {
51 missed.fetch_add(1, Ordering::Relaxed);
52 continue;
53 }
54
55 match tokio::time::timeout(std::time::Duration::from_secs(5), sub.recv()).await {
56 Ok(Some(_)) => {}
57 _ => {
58 missed.fetch_add(1, Ordering::Relaxed);
59 }
60 }
61 // Drop here, racing every other task's subscribe.
62 }
63 }));
64 }
65
66 for t in tasks {
67 t.await.unwrap();
68 }
69
70 assert_eq!(
71 missed.load(Ordering::Relaxed),
72 0,
73 "a live subscriber missed a publish, which means a prune raced a subscribe"
74 );
75 assert_eq!(hub.connection_count(), 0);
76 assert_eq!(hub.room_count(), 0, "every room channel was pruned");
77 }
78
79 /// Connection accounting must survive concurrent acquire and release, including
80 /// the rejected path. A counter that drifts upward silently shrinks the budget
81 /// until the process refuses connections it has room for.
82 #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
83 async fn connection_counters_do_not_drift_under_contention() {
84 let limits = HubLimits {
85 max_connections_per_user: 2,
86 max_connections_total: 16,
87 room_buffer: 8,
88 };
89 let hub = Hub::new(limits);
90 let rooms: Vec<RoomId> = (0..4).map(|_| RoomId(Uuid::new_v4())).collect();
91 let users: Vec<UserId> = (0..4).map(|_| UserId(Uuid::new_v4())).collect();
92
93 let mut tasks = Vec::new();
94 for t in 0..16 {
95 let hub = hub.clone();
96 let room = rooms[t % rooms.len()];
97 let user = users[t % users.len()];
98 tasks.push(tokio::spawn(async move {
99 for _ in 0..300 {
100 // Most of these are rejected by one budget or the other. Every
101 // rejection must leave both counters exactly as it found them.
102 if let Ok(sub) = hub.subscribe(room, user) {
103 tokio::task::yield_now().await;
104 drop(sub);
105 }
106 }
107 }));
108 }
109
110 for t in tasks {
111 t.await.unwrap();
112 }
113
114 assert_eq!(
115 hub.connection_count(),
116 0,
117 "global counter drifted; some path incremented without releasing"
118 );
119 assert_eq!(hub.room_count(), 0, "room channels leaked");
120
121 // The budget must be fully available again, which it is not if the counter
122 // drifted even by one.
123 let mut held = Vec::new();
124 for i in 0..limits.max_connections_total {
125 held.push(
126 hub.subscribe(rooms[i % rooms.len()], UserId(Uuid::new_v4()))
127 .expect("full budget available after churn"),
128 );
129 }
130 assert_eq!(hub.connection_count(), limits.max_connections_total);
131 }
132
133 /// Publishes to a room must reach every listener while others come and go.
134 #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
135 async fn a_stable_listener_receives_everything_through_churn() {
136 let hub = Hub::new(HubLimits {
137 max_connections_per_user: 512,
138 max_connections_total: 4096,
139 room_buffer: 4096,
140 });
141 let room = RoomId(Uuid::new_v4());
142
143 let mut stable = hub.subscribe(room, UserId(Uuid::new_v4())).unwrap();
144
145 let churn = {
146 let hub = hub.clone();
147 tokio::spawn(async move {
148 for _ in 0..500 {
149 let s = hub.subscribe(room, UserId(Uuid::new_v4())).unwrap();
150 tokio::task::yield_now().await;
151 drop(s);
152 }
153 })
154 };
155
156 let publisher = {
157 let hub = hub.clone();
158 tokio::spawn(async move {
159 for i in 0..500 {
160 hub.publish(room, event(i));
161 tokio::task::yield_now().await;
162 }
163 })
164 };
165
166 let mut received = 0;
167 while received < 500 {
168 match tokio::time::timeout(std::time::Duration::from_secs(5), stable.recv()).await {
169 Ok(Some(ChatEvent::Gap)) => panic!("buffer was sized to make a gap impossible here"),
170 Ok(Some(_)) => received += 1,
171 Ok(None) => panic!("stable listener's room was closed out from under it"),
172 Err(e) => panic!("stable listener stopped receiving after {received} events: {e:?}"),
173 }
174 }
175
176 churn.await.unwrap();
177 publisher.await.unwrap();
178 assert_eq!(received, 500);
179 }
180