Skip to main content

max / makenotwork

11.0 KB · 321 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 author: None,
128 }
129 }
130
131 fn ids(events: &[ChatEvent]) -> Vec<i64> {
132 events
133 .iter()
134 .filter_map(|e| match e {
135 ChatEvent::Message(m) => Some(m.id.0),
136 _ => None,
137 })
138 .collect()
139 }
140
141 async fn drain(stream: &mut ChatStream, n: usize) -> Vec<ChatEvent> {
142 let mut out = Vec::new();
143 for _ in 0..n {
144 match tokio::time::timeout(std::time::Duration::from_secs(5), stream.next()).await {
145 Ok(Some(e)) => out.push(e),
146 _ => break,
147 }
148 }
149 out
150 }
151
152 #[tokio::test]
153 async fn backlog_is_served_before_live() {
154 let hub = Hub::new(HubLimits::default());
155 let room = RoomId(Uuid::new_v4());
156
157 let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async {
158 Ok(vec![message(room, 1), message(room, 2)])
159 })
160 .await
161 .unwrap();
162
163 hub.publish(room, ChatEvent::Message(message(room, 3)));
164
165 assert_eq!(ids(&drain(&mut stream, 3).await), vec![1, 2, 3]);
166 }
167
168 #[tokio::test]
169 async fn a_message_published_during_the_fetch_is_not_lost() {
170 // The failure this exists for: fetch-then-subscribe drops anything sent
171 // in the window between the two.
172 let hub = Hub::new(HubLimits::default());
173 let room = RoomId(Uuid::new_v4());
174 let hub_for_fetch = hub.clone();
175
176 let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async move {
177 // Published while the fetch is in flight, and deliberately absent
178 // from what the fetch returns, exactly as a real query would miss a
179 // row committed after it ran.
180 hub_for_fetch.publish(room, ChatEvent::Message(message(room, 7)));
181 Ok(vec![message(room, 5), message(room, 6)])
182 })
183 .await
184 .unwrap();
185
186 assert_eq!(
187 ids(&drain(&mut stream, 3).await),
188 vec![5, 6, 7],
189 "the message sent during the fetch must still arrive"
190 );
191 }
192
193 #[tokio::test]
194 async fn the_overlap_does_not_duplicate() {
195 // The other side of subscribe-first: a message can appear in both the
196 // backlog and the live feed. It must be served once.
197 let hub = Hub::new(HubLimits::default());
198 let room = RoomId(Uuid::new_v4());
199 let hub_for_fetch = hub.clone();
200
201 let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async move {
202 hub_for_fetch.publish(room, ChatEvent::Message(message(room, 2)));
203 hub_for_fetch.publish(room, ChatEvent::Message(message(room, 3)));
204 // The fetch sees both, because they committed before it ran.
205 Ok(vec![message(room, 1), message(room, 2), message(room, 3)])
206 })
207 .await
208 .unwrap();
209
210 hub.publish(room, ChatEvent::Message(message(room, 4)));
211
212 assert_eq!(ids(&drain(&mut stream, 4).await), vec![1, 2, 3, 4]);
213 }
214
215 #[tokio::test]
216 async fn deletes_in_the_overlap_are_not_suppressed() {
217 // A delete carries no id that can be compared against the high-water
218 // mark, and it refers to a message the client already has. Dropping it
219 // would leave a deleted message on screen until reload.
220 let hub = Hub::new(HubLimits::default());
221 let room = RoomId(Uuid::new_v4());
222 let hub_for_fetch = hub.clone();
223
224 let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async move {
225 hub_for_fetch.publish(room, ChatEvent::Delete { id: MessageId(1) });
226 Ok(vec![message(room, 1), message(room, 2)])
227 })
228 .await
229 .unwrap();
230
231 let events = drain(&mut stream, 3).await;
232 assert!(
233 events
234 .iter()
235 .any(|e| matches!(e, ChatEvent::Delete { id: MessageId(1) })),
236 "delete during the overlap must pass through: {events:?}"
237 );
238 }
239
240 #[tokio::test]
241 async fn the_cursor_reaches_the_fetch() {
242 let hub = Hub::new(HubLimits::default());
243 let room = RoomId(Uuid::new_v4());
244 let seen = Arc::new(AtomicUsize::new(0));
245 let seen_clone = seen.clone();
246
247 let _stream = ChatStream::open(
248 &hub,
249 room,
250 UserId(Uuid::nil()),
251 Some(MessageId(42)),
252 move |after| async move {
253 assert_eq!(after, Some(MessageId(42)));
254 seen_clone.fetch_add(1, Ordering::SeqCst);
255 Ok(vec![])
256 },
257 )
258 .await
259 .unwrap();
260
261 assert_eq!(seen.load(Ordering::SeqCst), 1);
262 }
263
264 #[tokio::test]
265 async fn an_empty_backlog_goes_straight_to_live() {
266 let hub = Hub::new(HubLimits::default());
267 let room = RoomId(Uuid::new_v4());
268
269 let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async {
270 Ok(vec![])
271 })
272 .await
273 .unwrap();
274
275 hub.publish(room, ChatEvent::Message(message(room, 1)));
276 assert_eq!(ids(&drain(&mut stream, 1).await), vec![1]);
277 }
278
279 #[tokio::test]
280 async fn a_failing_fetch_releases_the_connection_slot() {
281 let hub = Hub::new(HubLimits::default());
282 let room = RoomId(Uuid::new_v4());
283
284 let result = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async {
285 Err(ChatError::MessageEmpty)
286 })
287 .await;
288
289 assert!(result.is_err());
290 assert_eq!(
291 hub.connection_count(),
292 0,
293 "the subscription taken before the fetch must be released when it fails"
294 );
295 assert_eq!(hub.room_count(), 0);
296 }
297
298 #[tokio::test]
299 async fn the_nonce_survives_the_seam() {
300 // The sender's own optimistic message must arrive carrying its nonce, or
301 // the client cannot reconcile and shows it twice.
302 let hub = Hub::new(HubLimits::default());
303 let room = RoomId(Uuid::new_v4());
304
305 let mut stream = ChatStream::open(&hub, room, UserId(Uuid::nil()), None, |_| async {
306 Ok(vec![])
307 })
308 .await
309 .unwrap();
310
311 let mut m = message(room, 1);
312 m.nonce = Nonce::parse("abc");
313 hub.publish(room, ChatEvent::Message(m));
314
315 match drain(&mut stream, 1).await.first() {
316 Some(ChatEvent::Message(m)) => assert_eq!(m.nonce, Nonce::parse("abc")),
317 other => panic!("expected a message carrying its nonce, got {other:?}"),
318 }
319 }
320 }
321