Skip to main content

max / makenotwork

10.9 KB · 320 lines History Blame Raw
1 //! Backlog replay joined to the live feed.
2 //!
3 //! # The ordering that matters
4 //!
5 //! Every deploy is a symlink swap and a restart with no connection draining, so
6 //! every client reconnects and asks to resume from a cursor. Serving that means
7 //! reading stored messages and then following the live feed, and the order those
8 //! two things happen in is the whole correctness question.
9 //!
10 //! Fetch the backlog first and then subscribe, and every message sent between the
11 //! two is lost: it is too new to be in the query result and too old to be in the
12 //! channel. Nothing errors, the client just has a hole exactly where it was busy
13 //! reconnecting, which is the worst possible moment.
14 //!
15 //! So [`ChatStream::open`] subscribes first, then fetches. That inverts the
16 //! problem into an overlap instead of a gap, and an overlap is fixable: anything
17 //! the live feed replays that the backlog already covered is dropped by id. The
18 //! host cannot get this wrong from the outside, because it hands over a fetch
19 //! closure rather than a finished backlog.
20
21 use crate::error::ChatError;
22 use crate::event::ChatEvent;
23 use crate::hub::{Hub, Subscription};
24 use crate::ids::{MessageId, RoomId, UserId};
25 use crate::message::Message;
26
27 /// Backlog replay followed by the live feed, deduplicated across the seam.
28 pub struct ChatStream {
29 backlog: std::vec::IntoIter<Message>,
30 live: Subscription,
31 /// Highest id served from the backlog. Live messages at or below it are
32 /// duplicates from the overlap window and are dropped.
33 high_water: Option<MessageId>,
34 }
35
36 impl ChatStream {
37 /// Subscribe, then fetch the backlog, then serve one followed by the other.
38 ///
39 /// `fetch` receives the client's cursor: `Some(id)` to resume after a known
40 /// message, `None` for a fresh connection, where the host should return
41 /// whatever recent window it wants a new arrival to see. It is called after
42 /// the subscription is live, which is the point.
43 ///
44 /// The host is expected to return messages in ascending id order. If it does
45 /// not, the seam still cannot duplicate (the high-water mark is taken from
46 /// the maximum, not the last), but the client will see the backlog in
47 /// whatever order it was given.
48 ///
49 /// # Errors
50 ///
51 /// [`ChatError::ConnectionLimit`] if the hub is at capacity, or whatever
52 /// `fetch` fails with.
53 pub async fn open<F, Fut>(
54 hub: &Hub,
55 room: RoomId,
56 user: UserId,
57 after: Option<MessageId>,
58 fetch: F,
59 ) -> Result<Self, ChatError>
60 where
61 F: FnOnce(Option<MessageId>) -> Fut,
62 Fut: Future<Output = Result<Vec<Message>, ChatError>>,
63 {
64 // Subscribe first. Everything published from here on is buffered for us
65 // while the fetch runs, so the two sources overlap rather than gap.
66 let live = hub.subscribe(room, user)?;
67
68 let backlog = fetch(after).await?;
69 let high_water = backlog.iter().map(|m| m.id).max();
70
71 Ok(Self {
72 backlog: backlog.into_iter(),
73 live,
74 high_water,
75 })
76 }
77
78 /// Next event, or `None` when the room is gone.
79 ///
80 /// Serves the backlog to exhaustion first, then the live feed.
81 pub async fn next(&mut self) -> Option<ChatEvent> {
82 if let Some(message) = self.backlog.next() {
83 return Some(ChatEvent::Message(message));
84 }
85
86 loop {
87 let event = self.live.recv().await?;
88
89 // Drop only what the backlog already served. Deletes and purges pass
90 // through regardless: a delete that lands during the overlap refers
91 // to a message the client has, so suppressing it would leave that
92 // message on screen forever.
93 if let ChatEvent::Message(m) = &event
94 && let Some(high) = self.high_water
95 && m.id <= high
96 {
97 continue;
98 }
99
100 return Some(event);
101 }
102 }
103
104 /// The room being served.
105 pub fn room(&self) -> RoomId {
106 self.live.room()
107 }
108 }
109
110 #[cfg(test)]
111 mod tests {
112 use super::*;
113 use crate::hub::HubLimits;
114 use crate::ids::Nonce;
115 use std::sync::Arc;
116 use std::sync::atomic::{AtomicUsize, Ordering};
117 use uuid::Uuid;
118
119 fn message(room: RoomId, id: i64) -> Message {
120 Message {
121 id: MessageId(id),
122 room_id: room,
123 author_id: UserId(Uuid::nil()),
124 body_html: format!("m{id}"),
125 created_at: 0,
126 nonce: None,
127 }
128 }
129
130 fn ids(events: &[ChatEvent]) -> Vec<i64> {
131 events
132 .iter()
133 .filter_map(|e| match e {
134 ChatEvent::Message(m) => Some(m.id.0),
135 _ => None,
136 })
137 .collect()
138 }
139
140 async fn drain(stream: &mut ChatStream, n: usize) -> Vec<ChatEvent> {
141 let mut out = Vec::new();
142 for _ in 0..n {
143 match tokio::time::timeout(std::time::Duration::from_secs(5), stream.next()).await {
144 Ok(Some(e)) => out.push(e),
145 _ => break,
146 }
147 }
148 out
149 }
150
151 #[tokio::test]
152 async fn backlog_is_served_before_live() {
153 let hub = Hub::new(HubLimits::default());
154 let room = RoomId(Uuid::new_v4());
155
156 let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async {
157 Ok(vec![message(room, 1), message(room, 2)])
158 })
159 .await
160 .unwrap();
161
162 hub.publish(room, ChatEvent::Message(message(room, 3)));
163
164 assert_eq!(ids(&drain(&mut stream, 3).await), vec![1, 2, 3]);
165 }
166
167 #[tokio::test]
168 async fn a_message_published_during_the_fetch_is_not_lost() {
169 // The failure this exists for: fetch-then-subscribe drops anything sent
170 // in the window between the two.
171 let hub = Hub::new(HubLimits::default());
172 let room = RoomId(Uuid::new_v4());
173 let hub_for_fetch = hub.clone();
174
175 let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async move {
176 // Published while the fetch is in flight, and deliberately absent
177 // from what the fetch returns, exactly as a real query would miss a
178 // row committed after it ran.
179 hub_for_fetch.publish(room, ChatEvent::Message(message(room, 7)));
180 Ok(vec![message(room, 5), message(room, 6)])
181 })
182 .await
183 .unwrap();
184
185 assert_eq!(
186 ids(&drain(&mut stream, 3).await),
187 vec![5, 6, 7],
188 "the message sent during the fetch must still arrive"
189 );
190 }
191
192 #[tokio::test]
193 async fn the_overlap_does_not_duplicate() {
194 // The other side of subscribe-first: a message can appear in both the
195 // backlog and the live feed. It must be served once.
196 let hub = Hub::new(HubLimits::default());
197 let room = RoomId(Uuid::new_v4());
198 let hub_for_fetch = hub.clone();
199
200 let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async move {
201 hub_for_fetch.publish(room, ChatEvent::Message(message(room, 2)));
202 hub_for_fetch.publish(room, ChatEvent::Message(message(room, 3)));
203 // The fetch sees both, because they committed before it ran.
204 Ok(vec![message(room, 1), message(room, 2), message(room, 3)])
205 })
206 .await
207 .unwrap();
208
209 hub.publish(room, ChatEvent::Message(message(room, 4)));
210
211 assert_eq!(ids(&drain(&mut stream, 4).await), vec![1, 2, 3, 4]);
212 }
213
214 #[tokio::test]
215 async fn deletes_in_the_overlap_are_not_suppressed() {
216 // A delete carries no id that can be compared against the high-water
217 // mark, and it refers to a message the client already has. Dropping it
218 // would leave a deleted message on screen until reload.
219 let hub = Hub::new(HubLimits::default());
220 let room = RoomId(Uuid::new_v4());
221 let hub_for_fetch = hub.clone();
222
223 let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async move {
224 hub_for_fetch.publish(room, ChatEvent::Delete { id: MessageId(1) });
225 Ok(vec![message(room, 1), message(room, 2)])
226 })
227 .await
228 .unwrap();
229
230 let events = drain(&mut stream, 3).await;
231 assert!(
232 events
233 .iter()
234 .any(|e| matches!(e, ChatEvent::Delete { id: MessageId(1) })),
235 "delete during the overlap must pass through: {events:?}"
236 );
237 }
238
239 #[tokio::test]
240 async fn the_cursor_reaches_the_fetch() {
241 let hub = Hub::new(HubLimits::default());
242 let room = RoomId(Uuid::new_v4());
243 let seen = Arc::new(AtomicUsize::new(0));
244 let seen_clone = seen.clone();
245
246 let _stream = ChatStream::open(
247 &hub,
248 room,
249 UserId(Uuid::nil()),
250 Some(MessageId(42)),
251 move |after| async move {
252 assert_eq!(after, Some(MessageId(42)));
253 seen_clone.fetch_add(1, Ordering::SeqCst);
254 Ok(vec![])
255 },
256 )
257 .await
258 .unwrap();
259
260 assert_eq!(seen.load(Ordering::SeqCst), 1);
261 }
262
263 #[tokio::test]
264 async fn an_empty_backlog_goes_straight_to_live() {
265 let hub = Hub::new(HubLimits::default());
266 let room = RoomId(Uuid::new_v4());
267
268 let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async {
269 Ok(vec![])
270 })
271 .await
272 .unwrap();
273
274 hub.publish(room, ChatEvent::Message(message(room, 1)));
275 assert_eq!(ids(&drain(&mut stream, 1).await), vec![1]);
276 }
277
278 #[tokio::test]
279 async fn a_failing_fetch_releases_the_connection_slot() {
280 let hub = Hub::new(HubLimits::default());
281 let room = RoomId(Uuid::new_v4());
282
283 let result = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async {
284 Err(ChatError::MessageEmpty)
285 })
286 .await;
287
288 assert!(result.is_err());
289 assert_eq!(
290 hub.connection_count(),
291 0,
292 "the subscription taken before the fetch must be released when it fails"
293 );
294 assert_eq!(hub.room_count(), 0);
295 }
296
297 #[tokio::test]
298 async fn the_nonce_survives_the_seam() {
299 // The sender's own optimistic message must arrive carrying its nonce, or
300 // the client cannot reconcile and shows it twice.
301 let hub = Hub::new(HubLimits::default());
302 let room = RoomId(Uuid::new_v4());
303
304 let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async {
305 Ok(vec![])
306 })
307 .await
308 .unwrap();
309
310 let mut m = message(room, 1);
311 m.nonce = Nonce::parse("abc");
312 hub.publish(room, ChatEvent::Message(m));
313
314 match drain(&mut stream, 1).await.first() {
315 Some(ChatEvent::Message(m)) => assert_eq!(m.nonce, Nonce::parse("abc")),
316 other => panic!("expected a message carrying its nonce, got {other:?}"),
317 }
318 }
319 }
320