Skip to main content

max / makenotwork

14.6 KB · 466 lines History Blame Raw
1 //! The moderation path.
2 //!
3 //! Same discipline as [`crate::send`], for the same reason: the order is a
4 //! correctness property, so it lives here once rather than in each host.
5 //! **Mutate durably, then publish.** Broadcasting a removal the database
6 //! refused produces a room where a message is gone for everyone present and
7 //! back for anyone who reloads, which is the deletion mirror of the phantom
8 //! message the send path is ordered to avoid.
9 //!
10 //! # Why a ban is one event
11 //!
12 //! A ban removes every message its target wrote in the retention window, and
13 //! that is a burst with no useful upper bound: the whole point of banning
14 //! someone is usually that they said a great many things. Fanning it out as one
15 //! [`ChatEvent::Delete`] per message would push that burst through every open
16 //! connection in the room at once, and the room buffer is finite, so past a few
17 //! hundred messages the deletes would overflow it and every listener would take
18 //! a [`ChatEvent::Gap`] and re-fetch the backlog. A moderation action that
19 //! reliably triggers a stampede is one moderators learn not to use.
20 //!
21 //! So the removal is a single statement keyed by (room, author) and a single
22 //! [`ChatEvent::Purge`], which the client applies by dropping everything it
23 //! holds from that author. Cost is constant in the number of messages.
24
25 use crate::error::ChatError;
26 use crate::event::ChatEvent;
27 use crate::hub::Hub;
28 use crate::ids::{MessageId, UserId};
29 use crate::room::Room;
30 use crate::traits::ChatModeration;
31
32 /// Remove one message, then tell the room.
33 ///
34 /// The host's [`ChatModeration::delete_message`] is responsible for proving
35 /// `actor` is entitled to this message, which is why the entitlement check is
36 /// not duplicated here: it needs the row, and only the host can read it.
37 pub(crate) async fn delete_message<M>(
38 hub: &Hub,
39 moderation: &M,
40 actor: UserId,
41 room: &Room,
42 message: MessageId,
43 ) -> Result<(), ChatError>
44 where
45 M: ChatModeration + ?Sized,
46 {
47 moderation.delete_message(actor, room, message).await?;
48 hub.publish(room.id, ChatEvent::Delete { id: message });
49 Ok(())
50 }
51
52 /// Remove every message by `target` in the room, then tell the room once.
53 ///
54 /// Returns how many rows went.
55 pub(crate) async fn purge_user<M>(
56 hub: &Hub,
57 moderation: &M,
58 actor: UserId,
59 room: &Room,
60 target: UserId,
61 ) -> Result<u64, ChatError>
62 where
63 M: ChatModeration + ?Sized,
64 {
65 let removed = moderation.purge_user(actor, room, target).await?;
66
67 // Broadcast even when nothing was removed. A connected client can be
68 // holding messages the retention sweep already deleted server-side, so
69 // "zero rows" does not mean "nothing on screen", and the event is one frame
70 // that clients apply idempotently.
71 hub.publish(room.id, ChatEvent::Purge { author_id: target });
72
73 Ok(removed)
74 }
75
76 /// Empty the room, then tell it once.
77 ///
78 /// Returns how many rows went. Same ordering rule as every other removal here:
79 /// a wipe the database refused must not clear anyone's screen.
80 pub(crate) async fn wipe_room<M>(
81 hub: &Hub,
82 moderation: &M,
83 actor: UserId,
84 room: &Room,
85 ) -> Result<u64, ChatError>
86 where
87 M: ChatModeration + ?Sized,
88 {
89 let removed = moderation.wipe_room(actor, room).await?;
90
91 // Announced even at zero rows, for the reason `purge_user` gives: a
92 // connected client can be holding messages the sweep already took, so an
93 // empty table does not mean an empty room on screen.
94 hub.publish(room.id, ChatEvent::Wipe);
95
96 Ok(removed)
97 }
98
99 /// Ban `target` and remove their backlog in one action.
100 ///
101 /// Returns how many messages the ban removed.
102 ///
103 /// The ban lands first. If the purge then fails, the caller gets the error and
104 /// the ban still stands, which is the right way round: a banned user with
105 /// messages still up is a cleanup job, while a purged user who can still type
106 /// is the incident continuing.
107 pub(crate) async fn ban_user<M>(
108 hub: &Hub,
109 moderation: &M,
110 actor: UserId,
111 room: &Room,
112 target: UserId,
113 reason: Option<&str>,
114 ) -> Result<u64, ChatError>
115 where
116 M: ChatModeration + ?Sized,
117 {
118 moderation.ban_user(actor, room, target, reason).await?;
119 purge_user(hub, moderation, actor, room, target).await
120 }
121
122 /// Stop `target` from sending for `duration`.
123 ///
124 /// Publishes nothing, deliberately. A timeout removes no content, so there is
125 /// nothing for a connected client to re-render, and reads are explicitly
126 /// unaffected. Announcing it to the room would turn a quiet moderation action
127 /// into a public one and hand every listener an event whose only content is
128 /// somebody else's punishment. The person timed out finds out when their next
129 /// send is refused with [`DenyReason::Muted`](crate::DenyReason::Muted).
130 pub(crate) async fn timeout_user<M>(
131 moderation: &M,
132 actor: UserId,
133 room: &Room,
134 target: UserId,
135 duration: std::time::Duration,
136 ) -> Result<(), ChatError>
137 where
138 M: ChatModeration + ?Sized,
139 {
140 moderation.timeout_user(actor, room, target, duration).await
141 }
142
143 #[cfg(test)]
144 mod tests {
145 use super::*;
146 use crate::hub::HubLimits;
147 use crate::ids::RoomId;
148 use crate::retention::Retention;
149 use crate::room::RoomState;
150 use async_trait::async_trait;
151 use std::sync::Mutex;
152 use std::sync::atomic::{AtomicBool, Ordering};
153 use std::time::Duration;
154 use uuid::Uuid;
155
156 /// Records what the host was asked to do, and can be told to fail.
157 #[derive(Default)]
158 struct Recorder {
159 calls: Mutex<Vec<String>>,
160 fail_ban: AtomicBool,
161 fail_purge: AtomicBool,
162 fail_delete: AtomicBool,
163 fail_wipe: AtomicBool,
164 purged_rows: u64,
165 wiped_rows: u64,
166 }
167
168 impl Recorder {
169 fn calls(&self) -> Vec<String> {
170 self.calls.lock().unwrap().clone()
171 }
172 fn note(&self, what: &str) {
173 self.calls.lock().unwrap().push(what.to_owned());
174 }
175 }
176
177 fn boom() -> ChatError {
178 ChatError::host(std::io::Error::other("db down"))
179 }
180
181 #[async_trait]
182 impl ChatModeration for Recorder {
183 async fn delete_message(
184 &self,
185 _actor: UserId,
186 _room: &Room,
187 message: MessageId,
188 ) -> Result<(), ChatError> {
189 self.note(&format!("delete {message}"));
190 if self.fail_delete.load(Ordering::SeqCst) {
191 return Err(boom());
192 }
193 Ok(())
194 }
195
196 async fn purge_user(
197 &self,
198 _actor: UserId,
199 _room: &Room,
200 _target: UserId,
201 ) -> Result<u64, ChatError> {
202 self.note("purge");
203 if self.fail_purge.load(Ordering::SeqCst) {
204 return Err(boom());
205 }
206 Ok(self.purged_rows)
207 }
208
209 async fn wipe_room(&self, _actor: UserId, _room: &Room) -> Result<u64, ChatError> {
210 self.note("wipe");
211 if self.fail_wipe.load(Ordering::SeqCst) {
212 return Err(boom());
213 }
214 Ok(self.wiped_rows)
215 }
216
217 async fn timeout_user(
218 &self,
219 _actor: UserId,
220 _room: &Room,
221 _target: UserId,
222 _duration: Duration,
223 ) -> Result<(), ChatError> {
224 self.note("timeout");
225 Ok(())
226 }
227
228 async fn ban_user(
229 &self,
230 _actor: UserId,
231 _room: &Room,
232 _target: UserId,
233 _reason: Option<&str>,
234 ) -> Result<(), ChatError> {
235 self.note("ban");
236 if self.fail_ban.load(Ordering::SeqCst) {
237 return Err(boom());
238 }
239 Ok(())
240 }
241
242 async fn log_action(
243 &self,
244 _actor: UserId,
245 _room_id: RoomId,
246 _action: &str,
247 _detail: Option<&str>,
248 ) -> Result<(), ChatError> {
249 Ok(())
250 }
251 }
252
253 fn room() -> Room {
254 Room {
255 id: RoomId(Uuid::new_v4()),
256 state: RoomState::Open,
257 retention: Retention::forum_default(),
258 }
259 }
260
261 fn user() -> UserId {
262 UserId(Uuid::new_v4())
263 }
264
265 /// Everything already queued on the room, without blocking.
266 ///
267 /// Polled rather than awaited because half these tests assert that *no*
268 /// event was published, and awaiting for that would mean picking a timeout
269 /// and trading a fast suite against a flaky one. Every event under test is
270 /// published before the assertion runs, so anything not queued by now is
271 /// genuinely absent.
272 fn drained(sub: &mut crate::hub::Subscription) -> Vec<ChatEvent> {
273 use std::task::{Context, Poll, Waker};
274
275 let mut out = Vec::new();
276 let mut cx = Context::from_waker(Waker::noop());
277 loop {
278 let mut recv = Box::pin(sub.recv());
279 match recv.as_mut().poll(&mut cx) {
280 Poll::Ready(Some(event)) => out.push(event),
281 Poll::Ready(None) | Poll::Pending => return out,
282 }
283 }
284 }
285
286 #[tokio::test]
287 async fn a_delete_reaches_the_room() {
288 let hub = Hub::new(HubLimits::default());
289 let r = room();
290 let mut listener = hub.subscribe(r.id, user()).unwrap();
291
292 delete_message(&hub, &Recorder::default(), user(), &r, MessageId(7))
293 .await
294 .unwrap();
295
296 assert_eq!(
297 drained(&mut listener),
298 vec![ChatEvent::Delete { id: MessageId(7) }]
299 );
300 }
301
302 #[tokio::test]
303 async fn a_failed_delete_publishes_nothing() {
304 // The mirror of the phantom-message failure: a message gone for
305 // everyone present and back on reload.
306 let hub = Hub::new(HubLimits::default());
307 let r = room();
308 let mut listener = hub.subscribe(r.id, user()).unwrap();
309
310 let host = Recorder::default();
311 host.fail_delete.store(true, Ordering::SeqCst);
312
313 assert!(
314 delete_message(&hub, &host, user(), &r, MessageId(7))
315 .await
316 .is_err()
317 );
318 assert!(
319 drained(&mut listener).is_empty(),
320 "a refused delete must not be broadcast"
321 );
322 }
323
324 #[tokio::test]
325 async fn a_ban_purges_and_emits_exactly_one_event() {
326 // The property the whole module exists for: constant event cost
327 // regardless of how much the banned user wrote.
328 let hub = Hub::new(HubLimits::default());
329 let r = room();
330 let target = user();
331 let mut listener = hub.subscribe(r.id, user()).unwrap();
332
333 let host = Recorder {
334 purged_rows: 4_000,
335 ..Default::default()
336 };
337
338 let removed = ban_user(&hub, &host, user(), &r, target, Some("spam"))
339 .await
340 .unwrap();
341
342 assert_eq!(removed, 4_000);
343 assert_eq!(host.calls(), vec!["ban", "purge"], "ban lands before purge");
344 assert_eq!(
345 drained(&mut listener),
346 vec![ChatEvent::Purge { author_id: target }],
347 "4000 messages removed, one event"
348 );
349 }
350
351 #[tokio::test]
352 async fn a_failed_ban_never_purges() {
353 let hub = Hub::new(HubLimits::default());
354 let r = room();
355 let mut listener = hub.subscribe(r.id, user()).unwrap();
356
357 let host = Recorder::default();
358 host.fail_ban.store(true, Ordering::SeqCst);
359
360 assert!(
361 ban_user(&hub, &host, user(), &r, user(), None)
362 .await
363 .is_err()
364 );
365 assert_eq!(host.calls(), vec!["ban"], "the purge must not have run");
366 assert!(drained(&mut listener).is_empty());
367 }
368
369 #[tokio::test]
370 async fn a_ban_whose_purge_fails_still_reports_the_error() {
371 let hub = Hub::new(HubLimits::default());
372 let r = room();
373 let mut listener = hub.subscribe(r.id, user()).unwrap();
374
375 let host = Recorder::default();
376 host.fail_purge.store(true, Ordering::SeqCst);
377
378 assert!(
379 ban_user(&hub, &host, user(), &r, user(), None)
380 .await
381 .is_err(),
382 "the caller must learn the backlog is still up"
383 );
384 assert_eq!(host.calls(), vec!["ban", "purge"]);
385 assert!(
386 drained(&mut listener).is_empty(),
387 "nothing was removed, so nothing is announced"
388 );
389 }
390
391 #[tokio::test]
392 async fn a_purge_that_removed_nothing_is_still_announced() {
393 // A client can hold messages the retention sweep already deleted, so
394 // zero rows does not mean nothing on screen.
395 let hub = Hub::new(HubLimits::default());
396 let r = room();
397 let target = user();
398 let mut listener = hub.subscribe(r.id, user()).unwrap();
399
400 let removed = purge_user(&hub, &Recorder::default(), user(), &r, target)
401 .await
402 .unwrap();
403
404 assert_eq!(removed, 0);
405 assert_eq!(
406 drained(&mut listener),
407 vec![ChatEvent::Purge { author_id: target }]
408 );
409 }
410
411 #[tokio::test]
412 async fn a_wipe_emits_one_event_however_much_it_removed() {
413 let hub = Hub::new(HubLimits::default());
414 let r = room();
415 let mut listener = hub.subscribe(r.id, user()).unwrap();
416
417 let host = Recorder {
418 wiped_rows: 12_000,
419 ..Default::default()
420 };
421
422 let removed = wipe_room(&hub, &host, user(), &r).await.unwrap();
423
424 assert_eq!(removed, 12_000);
425 assert_eq!(host.calls(), vec!["wipe"]);
426 assert_eq!(
427 drained(&mut listener),
428 vec![ChatEvent::Wipe],
429 "12000 messages removed, one event"
430 );
431 }
432
433 #[tokio::test]
434 async fn a_failed_wipe_leaves_the_room_on_screen() {
435 // Same rule as a failed delete: publishing a removal the database
436 // refused empties every screen in the room and refills them on reload.
437 let hub = Hub::new(HubLimits::default());
438 let r = room();
439 let mut listener = hub.subscribe(r.id, user()).unwrap();
440
441 let host = Recorder::default();
442 host.fail_wipe.store(true, Ordering::SeqCst);
443
444 assert!(wipe_room(&hub, &host, user(), &r).await.is_err());
445 assert!(drained(&mut listener).is_empty());
446 }
447
448 #[tokio::test]
449 async fn a_timeout_is_not_announced_to_the_room() {
450 let hub = Hub::new(HubLimits::default());
451 let r = room();
452 let mut listener = hub.subscribe(r.id, user()).unwrap();
453
454 let host = Recorder::default();
455 timeout_user(&host, user(), &r, user(), Duration::from_mins(5))
456 .await
457 .unwrap();
458
459 assert_eq!(host.calls(), vec!["timeout"]);
460 assert!(
461 drained(&mut listener).is_empty(),
462 "a timeout removes no content, so there is nothing to re-render"
463 );
464 }
465 }
466