Skip to main content

max / makenotwork

Add livechat send path behind a Chat front door send() sequences body validation, room state, rate limiting, authz, storage, and fan-out. The order is a security property, so it lives in the crate once rather than in each host. Rate limiting sits before authz so a caller hammering a room they cannot write to is throttled at memory speed instead of putting a query behind every attempt; the cost is that a refused sender still spends budget, which is the right trade. Publish happens only after the row is durable. The inverse feels faster and produces a room where a message is visible to everyone present, absent from the backlog, and gone on reload. Chat owns the hub and the limiter so a host holds one thing in application state instead of assembling three, and exposes sweep() with an interval that cannot turn bucket eviction into a bypass. This also drops send() from nine arguments to four. Design: wiki livechat-design Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-19 00:34 UTC
Commit: 53c2a00f12940e48697108ba87f453097ec13a18
Parent: abfb094
4 files changed, +595 insertions, -0 deletions
A shared/livechat/src/chat.rs +200
@@ -0,0 +1,200 @@
1 + //! The front door.
2 + //!
3 + //! [`Chat`] owns the crate's mutable state (the fan-out hub and the send
4 + //! budgets) so a host holds one thing in application state rather than
5 + //! assembling three and hoping it wires them together the same way twice.
6 +
7 + use std::time::{Duration, Instant};
8 +
9 + use crate::error::ChatError;
10 + use crate::hub::{Hub, HubLimits};
11 + use crate::ids::{MessageId, Nonce, RoomId, UserId};
12 + use crate::message::Message;
13 + use crate::rate_limit::{RateLimiter, RateLimits};
14 + use crate::room::Room;
15 + use crate::stream::ChatStream;
16 + use crate::traits::ChatAuthz;
17 +
18 + /// One send attempt.
19 + #[derive(Debug, Clone)]
20 + pub struct SendRequest<'a> {
21 + pub room: &'a Room,
22 + pub author: UserId,
23 + /// Unvalidated. Trimmed and length-checked on the way through.
24 + pub body: &'a str,
25 + /// The sender's optimistic-render nonce, echoed back so their own message
26 + /// reconciles instead of appearing twice.
27 + pub nonce: Option<Nonce>,
28 + /// Injected so the send path is testable without sleeping.
29 + pub now: Instant,
30 + }
31 +
32 + /// Chat state for one process.
33 + ///
34 + /// Cheap to clone: the hub is an `Arc` internally and the limiter is shared, so
35 + /// clones talk to the same rooms and the same budgets.
36 + pub struct Chat {
37 + hub: Hub,
38 + limiter: RateLimiter,
39 + }
40 +
41 + impl Chat {
42 + pub fn new(hub_limits: HubLimits, rate_limits: RateLimits) -> Self {
43 + Self {
44 + hub: Hub::new(hub_limits),
45 + limiter: RateLimiter::new(rate_limits),
46 + }
47 + }
48 +
49 + pub fn hub(&self) -> &Hub {
50 + &self.hub
51 + }
52 +
53 + pub fn limiter(&self) -> &RateLimiter {
54 + &self.limiter
55 + }
56 +
57 + /// Validate, authorize, store, and fan out one message.
58 + ///
59 + /// See [`crate::send`] for why the order of those steps is a security
60 + /// property rather than a preference.
61 + pub async fn send<A, F, Fut>(
62 + &self,
63 + authz: &A,
64 + req: SendRequest<'_>,
65 + store: F,
66 + ) -> Result<Message, ChatError>
67 + where
68 + A: ChatAuthz + ?Sized,
69 + F: FnOnce(crate::message::MessageBody) -> Fut,
70 + Fut: Future<Output = Result<Message, ChatError>>,
71 + {
72 + crate::send::send(&self.hub, &self.limiter, authz, req, store).await
73 + }
74 +
75 + /// Open a stream: backlog from `after`, then the live feed.
76 + ///
77 + /// Subscribes before fetching. See [`ChatStream`] for why that order is the
78 + /// whole correctness question.
79 + pub async fn subscribe<F, Fut>(
80 + &self,
81 + room: RoomId,
82 + user: UserId,
83 + after: Option<MessageId>,
84 + fetch: F,
85 + ) -> Result<ChatStream, ChatError>
86 + where
87 + F: FnOnce(Option<MessageId>) -> Fut,
88 + Fut: Future<Output = Result<Vec<Message>, ChatError>>,
89 + {
90 + ChatStream::open(&self.hub, room, user, after, fetch).await
91 + }
92 +
93 + /// Drop rate-limit buckets nobody has touched recently.
94 + ///
95 + /// The host must call this on a timer. Bucket state is keyed by (user, room),
96 + /// which is unbounded and attacker-influenced, and both consumers run under a
97 + /// hard 512M cgroup cap. Uses the shortest interval that cannot turn eviction
98 + /// into a way to skip the queue.
99 + ///
100 + /// Returns how many buckets were dropped.
101 + pub fn sweep(&self, now: Instant) -> usize {
102 + self.limiter
103 + .evict_idle(now, self.limiter.recommended_idle_timeout())
104 + }
105 +
106 + /// How often [`Chat::sweep`] wants to be called.
107 + pub fn sweep_interval(&self) -> Duration {
108 + self.limiter.recommended_idle_timeout()
109 + }
110 + }
111 +
112 + #[cfg(test)]
113 + mod tests {
114 + use super::*;
115 + use crate::room::RoomState;
116 + use crate::retention::Retention;
117 + use crate::traits::WriteAccess;
118 + use async_trait::async_trait;
119 + use uuid::Uuid;
120 +
121 + struct Yes;
122 +
123 + #[async_trait]
124 + impl ChatAuthz for Yes {
125 + async fn can_read(&self, _v: Option<UserId>, _r: &Room) -> Result<bool, ChatError> {
126 + Ok(true)
127 + }
128 + async fn can_write(&self, _u: UserId, _r: &Room) -> Result<WriteAccess, ChatError> {
129 + Ok(WriteAccess::Allow)
130 + }
131 + async fn is_moderator(&self, _u: UserId, _r: &Room) -> Result<bool, ChatError> {
132 + Ok(false)
133 + }
134 + }
135 +
136 + fn room() -> Room {
137 + Room {
138 + id: RoomId(Uuid::new_v4()),
139 + state: RoomState::Open,
140 + retention: Retention::forum_default(),
141 + }
142 + }
143 +
144 + #[tokio::test]
145 + async fn send_and_subscribe_share_the_same_rooms() {
146 + let chat = Chat::new(HubLimits::default(), RateLimits::default());
147 + let r = room();
148 + let author = UserId(Uuid::new_v4());
149 +
150 + let mut stream = chat
151 + .subscribe(r.id, UserId(Uuid::new_v4()), None, |_| async { Ok(vec![]) })
152 + .await
153 + .unwrap();
154 +
155 + chat.send(
156 + &Yes,
157 + SendRequest {
158 + room: &r,
159 + author,
160 + body: "hi",
161 + nonce: None,
162 + now: Instant::now(),
163 + },
164 + |body| async move {
165 + Ok(Message {
166 + id: MessageId(1),
167 + room_id: r.id,
168 + author_id: author,
169 + body_html: body.as_str().to_owned(),
170 + created_at: 0,
171 + nonce: None,
172 + })
173 + },
174 + )
175 + .await
176 + .unwrap();
177 +
178 + assert!(
179 + matches!(stream.next().await, Some(crate::event::ChatEvent::Message(_))),
180 + "a message sent through Chat reaches a stream opened through Chat"
181 + );
182 + }
183 +
184 + #[test]
185 + fn sweep_uses_an_interval_that_is_not_a_bypass() {
186 + let chat = Chat::new(
187 + HubLimits::default(),
188 + RateLimits {
189 + burst: 5,
190 + sustain_per_min: 60,
191 + },
192 + );
193 + assert_eq!(chat.sweep_interval(), Duration::from_secs(5));
194 +
195 + let t0 = Instant::now();
196 + chat.limiter().check(UserId(Uuid::nil()), RoomId(Uuid::nil()), t0);
197 + assert_eq!(chat.sweep(t0), 0, "a fresh bucket is not idle");
198 + assert_eq!(chat.sweep(t0 + Duration::from_secs(10)), 1);
199 + }
200 + }
M shared/livechat/src/error.rs +5
@@ -60,6 +60,11 @@ pub enum ChatError {
60 60 #[error("connection limit reached ({0})")]
61 61 ConnectionLimit(LimitScope),
62 62
63 + /// The sender did something the room refused. Distinct from a host error:
64 + /// this is expected traffic, not a fault, and should not be logged as one.
65 + #[error("send rejected: {0:?}")]
66 + Rejected(crate::send::SendRejection),
67 +
63 68 /// An error from the host app's trait impl: a query failure, a session
64 69 /// lookup, anything the crate has no vocabulary for.
65 70 #[error("host error: {0}")]
M shared/livechat/src/lib.rs +4
@@ -37,6 +37,7 @@
37 37 //! unreviewed user input at chat volume, which is a much hotter SSRF surface than
38 38 //! the forum-post path it would borrow.
39 39
40 + mod chat;
40 41 mod error;
41 42 mod event;
42 43 mod hub;
@@ -45,11 +46,13 @@ mod message;
45 46 mod rate_limit;
46 47 mod retention;
47 48 mod room;
49 + mod send;
48 50 #[cfg(feature = "axum")]
49 51 mod sse;
50 52 mod stream;
51 53 mod traits;
52 54
55 + pub use chat::{Chat, SendRequest};
53 56 pub use error::{ChatError, LimitScope};
54 57 pub use event::ChatEvent;
55 58 pub use hub::{Hub, HubLimits, Subscription};
@@ -58,6 +61,7 @@ pub use message::{MAX_MESSAGE_LEN, Message, MessageBody};
58 61 pub use rate_limit::{RateDecision, RateLimiter, RateLimits};
59 62 pub use retention::{MAX_AGE_CEILING, MAX_MESSAGES_CEILING, Retention};
60 63 pub use room::{Room, RoomState};
64 + pub use send::SendRejection;
61 65 #[cfg(feature = "axum")]
62 66 pub use sse::{KEEPALIVE_INTERVAL, sse_response};
63 67 pub use stream::ChatStream;
A shared/livechat/src/send.rs +386
@@ -0,0 +1,386 @@
1 + //! The send path.
2 + //!
3 + //! Six things have to happen in a particular order, and the order is a security
4 + //! property rather than a style preference, so it lives here once instead of in
5 + //! each host.
6 + //!
7 + //! 1. **Validate the body.** Free, and rejects the largest class of junk.
8 + //! 2. **Check room state.** Free, and a closed or read-only room ends it.
9 + //! 3. **Take a rate token.** In-memory. Deliberately before authz, so a caller
10 + //! hammering a room they cannot write to is throttled at memory speed instead
11 + //! of putting a database query behind every attempt. The cost is that a
12 + //! refused sender still spends budget, which is the right trade: the budget
13 + //! exists to protect the room and the box, not to be fair to someone who is
14 + //! already being refused.
15 + //! 4. **Check authz.** The first step that touches the host's database.
16 + //! 5. **Store.** The host renders through docengine and assigns the id.
17 + //! 6. **Publish.** Only after the row is durable, so nobody sees a message that
18 + //! a failed insert means will not survive a reconnect.
19 + //!
20 + //! Getting 5 and 6 backwards is the interesting mistake. Publishing first makes
21 + //! chat feel faster and produces a room where a message is visible to everyone
22 + //! present, absent from the backlog, and gone for anyone who reloads.
23 +
24 + use crate::chat::SendRequest;
25 + use crate::error::ChatError;
26 + use crate::event::ChatEvent;
27 + use crate::hub::Hub;
28 + use crate::message::{Message, MessageBody};
29 + use crate::rate_limit::{RateDecision, RateLimiter};
30 + use crate::traits::{ChatAuthz, DenyReason, WriteAccess};
31 +
32 + /// Why a send was refused, distinguishing the cases a caller must respond to
33 + /// differently.
34 + #[derive(Debug, Clone, PartialEq, Eq)]
35 + pub enum SendRejection {
36 + /// The body did not survive validation.
37 + Invalid(String),
38 + /// The room is not accepting messages.
39 + RoomClosed,
40 + RoomReadOnly,
41 + /// Authz or rate limiting said no.
42 + Denied(DenyReason),
43 + }
44 +
45 + /// Validate, authorize, store, and fan out one message.
46 + ///
47 + /// Reached through [`crate::Chat::send`]; the module docs above explain the
48 + /// ordering it enforces.
49 + ///
50 + /// `store` receives the validated body and must render it (through docengine's
51 + /// chat preset, never raw) and insert it, returning the stored [`Message`]. It is
52 + /// called only after every check has passed.
53 + pub(crate) async fn send<A, F, Fut>(
54 + hub: &Hub,
55 + limiter: &RateLimiter,
56 + authz: &A,
57 + req: SendRequest<'_>,
58 + store: F,
59 + ) -> Result<Message, ChatError>
60 + where
61 + A: ChatAuthz + ?Sized,
62 + F: FnOnce(MessageBody) -> Fut,
63 + Fut: Future<Output = Result<Message, ChatError>>,
64 + {
65 + let SendRequest {
66 + room,
67 + author,
68 + body: raw_body,
69 + nonce,
70 + now,
71 + } = req;
72 +
73 + // 1. Free.
74 + let body = MessageBody::parse(raw_body)
75 + .map_err(|e| ChatError::Rejected(SendRejection::Invalid(e.to_string())))?;
76 +
77 + // 2. Free.
78 + if !room.state.allows_reads() {
79 + return Err(ChatError::Rejected(SendRejection::RoomClosed));
80 + }
81 + if !room.state.allows_writes() {
82 + return Err(ChatError::Rejected(SendRejection::RoomReadOnly));
83 + }
84 +
85 + // 3. In-memory, before anything that queries.
86 + if let RateDecision::Deny { retry_after } = limiter.check(author, room.id, now) {
87 + return Err(ChatError::Rejected(SendRejection::Denied(
88 + DenyReason::RateLimited { retry_after },
89 + )));
90 + }
91 +
92 + // 4. First database hit.
93 + if let WriteAccess::Deny(reason) = authz.can_write(author, room).await? {
94 + return Err(ChatError::Rejected(SendRejection::Denied(reason)));
95 + }
96 +
97 + // 5. Durable before visible.
98 + let mut stored = store(body).await?;
99 + stored.nonce = nonce;
100 +
101 + // 6. Fan out. A room with nobody in it is normal and not a failure.
102 + hub.publish(room.id, ChatEvent::Message(stored.clone()));
103 +
104 + Ok(stored)
105 + }
106 +
107 + #[cfg(test)]
108 + mod tests {
109 + use super::*;
110 + use crate::hub::HubLimits;
111 + use crate::ids::{MessageId, Nonce, RoomId, UserId};
112 + use crate::message::MAX_MESSAGE_LEN;
113 + use crate::rate_limit::RateLimits;
114 + use crate::retention::Retention;
115 + use crate::room::{Room, RoomState};
116 + use async_trait::async_trait;
117 + use std::sync::Arc;
118 + use std::sync::atomic::{AtomicUsize, Ordering};
119 + use std::time::Instant;
120 + use uuid::Uuid;
121 +
122 + struct Authz {
123 + decision: WriteAccess,
124 + calls: Arc<AtomicUsize>,
125 + }
126 +
127 + #[async_trait]
128 + impl ChatAuthz for Authz {
129 + async fn can_read(&self, _v: Option<UserId>, room: &Room) -> Result<bool, ChatError> {
130 + Ok(room.state.allows_reads())
131 + }
132 + async fn can_write(&self, _u: UserId, _r: &Room) -> Result<WriteAccess, ChatError> {
133 + self.calls.fetch_add(1, Ordering::SeqCst);
134 + Ok(self.decision.clone())
135 + }
136 + async fn is_moderator(&self, _u: UserId, _r: &Room) -> Result<bool, ChatError> {
137 + Ok(false)
138 + }
139 + }
140 +
141 + fn authz(decision: WriteAccess) -> (Authz, Arc<AtomicUsize>) {
142 + let calls = Arc::new(AtomicUsize::new(0));
143 + (
144 + Authz {
145 + decision,
146 + calls: calls.clone(),
147 + },
148 + calls,
149 + )
150 + }
151 +
152 + fn room(state: RoomState) -> Room {
153 + Room {
154 + id: RoomId(Uuid::new_v4()),
155 + state,
156 + retention: Retention::forum_default(),
157 + }
158 + }
159 +
160 + fn stored(room: RoomId, author: UserId, body: &MessageBody) -> Message {
161 + Message {
162 + id: MessageId(1),
163 + room_id: room,
164 + author_id: author,
165 + body_html: format!("<p>{}</p>", body.as_str()),
166 + created_at: 0,
167 + nonce: None,
168 + }
169 + }
170 +
171 + fn parts() -> (Hub, RateLimiter) {
172 + (
173 + Hub::new(HubLimits::default()),
174 + RateLimiter::new(RateLimits::default()),
175 + )
176 + }
177 +
178 + fn req<'a>(room: &'a Room, author: UserId, body: &'a str) -> SendRequest<'a> {
179 + SendRequest {
180 + room,
181 + author,
182 + body,
183 + nonce: None,
184 + now: Instant::now(),
185 + }
186 + }
187 +
188 + #[tokio::test]
189 + async fn a_good_message_is_stored_then_published() {
190 + let (hub, rl) = parts();
191 + let (a, _) = authz(WriteAccess::Allow);
192 + let r = room(RoomState::Open);
193 + let author = UserId(Uuid::new_v4());
194 + let mut listener = hub.subscribe(r.id, UserId(Uuid::new_v4())).unwrap();
195 +
196 + let out = send(
197 + &hub,
198 + &rl,
199 + &a,
200 + SendRequest {
201 + nonce: Nonce::parse("n1"),
202 + ..req(&r, author, " hello ")
203 + },
204 + |body| async move { Ok(stored(r.id, author, &body)) },
205 + )
206 + .await
207 + .unwrap();
208 +
209 + assert_eq!(out.body_html, "<p>hello</p>", "body was trimmed and rendered");
210 + assert_eq!(out.nonce, Nonce::parse("n1"));
211 +
212 + match listener.recv().await {
213 + Some(ChatEvent::Message(m)) => {
214 + assert_eq!(m.id, MessageId(1));
215 + assert_eq!(m.nonce, Nonce::parse("n1"), "sender can reconcile");
216 + }
217 + other => panic!("expected the message on the room, got {other:?}"),
218 + }
219 + }
220 +
221 + #[tokio::test]
222 + async fn nothing_is_published_when_the_store_fails() {
223 + // The ordering this protects: a message visible to everyone present but
224 + // absent from the backlog and gone on reload.
225 + let (hub, rl) = parts();
226 + let (a, _) = authz(WriteAccess::Allow);
227 + let r = room(RoomState::Open);
228 + let mut listener = hub.subscribe(r.id, UserId(Uuid::new_v4())).unwrap();
229 +
230 + let result = send(
231 + &hub,
232 + &rl,
233 + &a,
234 + req(&r, UserId(Uuid::new_v4()), "hi"),
235 + |_| async { Err(ChatError::host(std::io::Error::other("db down"))) },
236 + )
237 + .await;
238 +
239 + assert!(result.is_err());
240 + // Probe the room. If a phantom message had been published it would be
241 + // ahead of this in the queue.
242 + assert_eq!(hub.publish(r.id, ChatEvent::Gap), 1, "listener still live");
243 + assert!(
244 + matches!(listener.recv().await, Some(ChatEvent::Gap)),
245 + "a failed store must publish nothing"
246 + );
247 + }
248 +
249 + #[tokio::test]
250 + async fn an_empty_body_never_reaches_authz() {
251 + let (hub, rl) = parts();
252 + let (a, calls) = authz(WriteAccess::Allow);
253 + let r = room(RoomState::Open);
254 +
255 + let result = send(&hub, &rl, &a, req(&r, UserId(Uuid::new_v4()), " "), |_| async {
256 + panic!("store must not run")
257 + })
258 + .await;
259 +
260 + assert!(matches!(
261 + result,
262 + Err(ChatError::Rejected(SendRejection::Invalid(_)))
263 + ));
264 + assert_eq!(calls.load(Ordering::SeqCst), 0, "no query for a blank body");
265 + }
266 +
267 + #[tokio::test]
268 + async fn an_overlong_body_is_refused() {
269 + let (hub, rl) = parts();
270 + let (a, _) = authz(WriteAccess::Allow);
271 + let r = room(RoomState::Open);
272 + let long = "x".repeat(MAX_MESSAGE_LEN + 1);
273 +
274 + let result = send(&hub, &rl, &a, req(&r, UserId(Uuid::new_v4()), &long), |_| async {
275 + panic!("store must not run")
276 + })
277 + .await;
278 +
279 + assert!(matches!(
280 + result,
281 + Err(ChatError::Rejected(SendRejection::Invalid(_)))
282 + ));
283 + }
284 +
285 + #[tokio::test]
286 + async fn a_read_only_room_refuses_before_querying() {
287 + let (hub, rl) = parts();
288 + let (a, calls) = authz(WriteAccess::Allow);
289 + let r = room(RoomState::ReadOnly);
290 +
291 + let result = send(&hub, &rl, &a, req(&r, UserId(Uuid::new_v4()), "hi"), |_| async {
292 + panic!("store must not run")
293 + })
294 + .await;
295 +
296 + assert!(matches!(
297 + result,
298 + Err(ChatError::Rejected(SendRejection::RoomReadOnly))
299 + ));
300 + assert_eq!(calls.load(Ordering::SeqCst), 0);
301 + }
302 +
303 + #[tokio::test]
304 + async fn a_closed_room_refuses_as_closed_not_read_only() {
305 + let (hub, rl) = parts();
306 + let (a, _) = authz(WriteAccess::Allow);
307 + let r = room(RoomState::Closed);
308 +
309 + let result = send(&hub, &rl, &a, req(&r, UserId(Uuid::new_v4()), "hi"), |_| async {
310 + panic!("store must not run")
311 + })
312 + .await;
313 +
314 + assert!(matches!(
315 + result,
316 + Err(ChatError::Rejected(SendRejection::RoomClosed))
317 + ));
318 + }
319 +
320 + #[tokio::test]
321 + async fn a_denied_sender_gets_the_specific_reason() {
322 + let (hub, rl) = parts();
323 + let (a, _) = authz(WriteAccess::Deny(DenyReason::Muted));
324 + let r = room(RoomState::Open);
325 +
326 + let result = send(&hub, &rl, &a, req(&r, UserId(Uuid::new_v4()), "hi"), |_| async {
327 + panic!("store must not run")
328 + })
329 + .await;
330 +
331 + assert!(matches!(
332 + result,
333 + Err(ChatError::Rejected(SendRejection::Denied(DenyReason::Muted)))
334 + ));
335 + }
336 +
337 + #[tokio::test]
338 + async fn the_rate_limit_bites_before_authz_is_queried() {
339 + // A caller hammering a room is throttled in memory rather than putting a
340 + // query behind every attempt.
341 + let hub = Hub::new(HubLimits::default());
342 + let rl = RateLimiter::new(RateLimits {
343 + burst: 2,
344 + sustain_per_min: 60,
345 + });
346 + let (a, calls) = authz(WriteAccess::Allow);
347 + let r = room(RoomState::Open);
348 + let author = UserId(Uuid::new_v4());
349 + let t0 = Instant::now();
350 +
351 + let at = |body| SendRequest {
352 + room: &r,
353 + author,
354 + body,
355 + nonce: None,
356 + now: t0,
357 + };
358 +
359 + for _ in 0..2 {
360 + send(&hub, &rl, &a, at("hi"), |body| async move {
361 + Ok(stored(r.id, author, &body))
362 + })
363 + .await
364 + .unwrap();
365 + }
366 + assert_eq!(calls.load(Ordering::SeqCst), 2);
367 +
368 + for _ in 0..10 {
369 + let result = send(&hub, &rl, &a, at("hi"), |_| async {
370 + panic!("store must not run")
371 + })
372 + .await;
373 + assert!(matches!(
374 + result,
375 + Err(ChatError::Rejected(SendRejection::Denied(
376 + DenyReason::RateLimited { .. }
377 + )))
378 + ));
379 + }
380 + assert_eq!(
381 + calls.load(Ordering::SeqCst),
382 + 2,
383 + "throttled attempts must not reach the database"
384 + );
385 + }
386 + }