Skip to main content

max / makenotwork

12.4 KB · 396 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 /// Ban `target` and remove their backlog in one action.
77 ///
78 /// Returns how many messages the ban removed.
79 ///
80 /// The ban lands first. If the purge then fails, the caller gets the error and
81 /// the ban still stands, which is the right way round: a banned user with
82 /// messages still up is a cleanup job, while a purged user who can still type
83 /// is the incident continuing.
84 pub(crate) async fn ban_user<M>(
85 hub: &Hub,
86 moderation: &M,
87 actor: UserId,
88 room: &Room,
89 target: UserId,
90 reason: Option<&str>,
91 ) -> Result<u64, ChatError>
92 where
93 M: ChatModeration + ?Sized,
94 {
95 moderation.ban_user(actor, room, target, reason).await?;
96 purge_user(hub, moderation, actor, room, target).await
97 }
98
99 /// Stop `target` from sending for `duration`.
100 ///
101 /// Publishes nothing, deliberately. A timeout removes no content, so there is
102 /// nothing for a connected client to re-render, and reads are explicitly
103 /// unaffected. Announcing it to the room would turn a quiet moderation action
104 /// into a public one and hand every listener an event whose only content is
105 /// somebody else's punishment. The person timed out finds out when their next
106 /// send is refused with [`DenyReason::Muted`](crate::DenyReason::Muted).
107 pub(crate) async fn timeout_user<M>(
108 moderation: &M,
109 actor: UserId,
110 room: &Room,
111 target: UserId,
112 duration: std::time::Duration,
113 ) -> Result<(), ChatError>
114 where
115 M: ChatModeration + ?Sized,
116 {
117 moderation.timeout_user(actor, room, target, duration).await
118 }
119
120 #[cfg(test)]
121 mod tests {
122 use super::*;
123 use crate::hub::HubLimits;
124 use crate::ids::RoomId;
125 use crate::retention::Retention;
126 use crate::room::RoomState;
127 use async_trait::async_trait;
128 use std::sync::Mutex;
129 use std::sync::atomic::{AtomicBool, Ordering};
130 use std::time::Duration;
131 use uuid::Uuid;
132
133 /// Records what the host was asked to do, and can be told to fail.
134 #[derive(Default)]
135 struct Recorder {
136 calls: Mutex<Vec<String>>,
137 fail_ban: AtomicBool,
138 fail_purge: AtomicBool,
139 fail_delete: AtomicBool,
140 purged_rows: u64,
141 }
142
143 impl Recorder {
144 fn calls(&self) -> Vec<String> {
145 self.calls.lock().unwrap().clone()
146 }
147 fn note(&self, what: &str) {
148 self.calls.lock().unwrap().push(what.to_owned());
149 }
150 }
151
152 fn boom() -> ChatError {
153 ChatError::host(std::io::Error::other("db down"))
154 }
155
156 #[async_trait]
157 impl ChatModeration for Recorder {
158 async fn delete_message(
159 &self,
160 _actor: UserId,
161 _room: &Room,
162 message: MessageId,
163 ) -> Result<(), ChatError> {
164 self.note(&format!("delete {message}"));
165 if self.fail_delete.load(Ordering::SeqCst) {
166 return Err(boom());
167 }
168 Ok(())
169 }
170
171 async fn purge_user(
172 &self,
173 _actor: UserId,
174 _room: &Room,
175 _target: UserId,
176 ) -> Result<u64, ChatError> {
177 self.note("purge");
178 if self.fail_purge.load(Ordering::SeqCst) {
179 return Err(boom());
180 }
181 Ok(self.purged_rows)
182 }
183
184 async fn timeout_user(
185 &self,
186 _actor: UserId,
187 _room: &Room,
188 _target: UserId,
189 _duration: Duration,
190 ) -> Result<(), ChatError> {
191 self.note("timeout");
192 Ok(())
193 }
194
195 async fn ban_user(
196 &self,
197 _actor: UserId,
198 _room: &Room,
199 _target: UserId,
200 _reason: Option<&str>,
201 ) -> Result<(), ChatError> {
202 self.note("ban");
203 if self.fail_ban.load(Ordering::SeqCst) {
204 return Err(boom());
205 }
206 Ok(())
207 }
208
209 async fn log_action(
210 &self,
211 _actor: UserId,
212 _room_id: RoomId,
213 _action: &str,
214 _detail: Option<&str>,
215 ) -> Result<(), ChatError> {
216 Ok(())
217 }
218 }
219
220 fn room() -> Room {
221 Room {
222 id: RoomId(Uuid::new_v4()),
223 state: RoomState::Open,
224 retention: Retention::forum_default(),
225 }
226 }
227
228 fn user() -> UserId {
229 UserId(Uuid::new_v4())
230 }
231
232 /// Everything already queued on the room, without blocking.
233 ///
234 /// Polled rather than awaited because half these tests assert that *no*
235 /// event was published, and awaiting for that would mean picking a timeout
236 /// and trading a fast suite against a flaky one. Every event under test is
237 /// published before the assertion runs, so anything not queued by now is
238 /// genuinely absent.
239 fn drained(sub: &mut crate::hub::Subscription) -> Vec<ChatEvent> {
240 use std::task::{Context, Poll, Waker};
241
242 let mut out = Vec::new();
243 let mut cx = Context::from_waker(Waker::noop());
244 loop {
245 let mut recv = Box::pin(sub.recv());
246 match recv.as_mut().poll(&mut cx) {
247 Poll::Ready(Some(event)) => out.push(event),
248 Poll::Ready(None) | Poll::Pending => return out,
249 }
250 }
251 }
252
253 #[tokio::test]
254 async fn a_delete_reaches_the_room() {
255 let hub = Hub::new(HubLimits::default());
256 let r = room();
257 let mut listener = hub.subscribe(r.id, user()).unwrap();
258
259 delete_message(&hub, &Recorder::default(), user(), &r, MessageId(7))
260 .await
261 .unwrap();
262
263 assert_eq!(
264 drained(&mut listener),
265 vec![ChatEvent::Delete { id: MessageId(7) }]
266 );
267 }
268
269 #[tokio::test]
270 async fn a_failed_delete_publishes_nothing() {
271 // The mirror of the phantom-message failure: a message gone for
272 // everyone present and back on reload.
273 let hub = Hub::new(HubLimits::default());
274 let r = room();
275 let mut listener = hub.subscribe(r.id, user()).unwrap();
276
277 let host = Recorder::default();
278 host.fail_delete.store(true, Ordering::SeqCst);
279
280 assert!(
281 delete_message(&hub, &host, user(), &r, MessageId(7))
282 .await
283 .is_err()
284 );
285 assert!(
286 drained(&mut listener).is_empty(),
287 "a refused delete must not be broadcast"
288 );
289 }
290
291 #[tokio::test]
292 async fn a_ban_purges_and_emits_exactly_one_event() {
293 // The property the whole module exists for: constant event cost
294 // regardless of how much the banned user wrote.
295 let hub = Hub::new(HubLimits::default());
296 let r = room();
297 let target = user();
298 let mut listener = hub.subscribe(r.id, user()).unwrap();
299
300 let host = Recorder {
301 purged_rows: 4_000,
302 ..Default::default()
303 };
304
305 let removed = ban_user(&hub, &host, user(), &r, target, Some("spam"))
306 .await
307 .unwrap();
308
309 assert_eq!(removed, 4_000);
310 assert_eq!(host.calls(), vec!["ban", "purge"], "ban lands before purge");
311 assert_eq!(
312 drained(&mut listener),
313 vec![ChatEvent::Purge { author_id: target }],
314 "4000 messages removed, one event"
315 );
316 }
317
318 #[tokio::test]
319 async fn a_failed_ban_never_purges() {
320 let hub = Hub::new(HubLimits::default());
321 let r = room();
322 let mut listener = hub.subscribe(r.id, user()).unwrap();
323
324 let host = Recorder::default();
325 host.fail_ban.store(true, Ordering::SeqCst);
326
327 assert!(
328 ban_user(&hub, &host, user(), &r, user(), None)
329 .await
330 .is_err()
331 );
332 assert_eq!(host.calls(), vec!["ban"], "the purge must not have run");
333 assert!(drained(&mut listener).is_empty());
334 }
335
336 #[tokio::test]
337 async fn a_ban_whose_purge_fails_still_reports_the_error() {
338 let hub = Hub::new(HubLimits::default());
339 let r = room();
340 let mut listener = hub.subscribe(r.id, user()).unwrap();
341
342 let host = Recorder::default();
343 host.fail_purge.store(true, Ordering::SeqCst);
344
345 assert!(
346 ban_user(&hub, &host, user(), &r, user(), None)
347 .await
348 .is_err(),
349 "the caller must learn the backlog is still up"
350 );
351 assert_eq!(host.calls(), vec!["ban", "purge"]);
352 assert!(
353 drained(&mut listener).is_empty(),
354 "nothing was removed, so nothing is announced"
355 );
356 }
357
358 #[tokio::test]
359 async fn a_purge_that_removed_nothing_is_still_announced() {
360 // A client can hold messages the retention sweep already deleted, so
361 // zero rows does not mean nothing on screen.
362 let hub = Hub::new(HubLimits::default());
363 let r = room();
364 let target = user();
365 let mut listener = hub.subscribe(r.id, user()).unwrap();
366
367 let removed = purge_user(&hub, &Recorder::default(), user(), &r, target)
368 .await
369 .unwrap();
370
371 assert_eq!(removed, 0);
372 assert_eq!(
373 drained(&mut listener),
374 vec![ChatEvent::Purge { author_id: target }]
375 );
376 }
377
378 #[tokio::test]
379 async fn a_timeout_is_not_announced_to_the_room() {
380 let hub = Hub::new(HubLimits::default());
381 let r = room();
382 let mut listener = hub.subscribe(r.id, user()).unwrap();
383
384 let host = Recorder::default();
385 timeout_user(&host, user(), &r, user(), Duration::from_mins(5))
386 .await
387 .unwrap();
388
389 assert_eq!(host.calls(), vec!["timeout"]);
390 assert!(
391 drained(&mut listener).is_empty(),
392 "a timeout removes no content, so there is nothing to re-render"
393 );
394 }
395 }
396