Skip to main content

max / makenotwork

multithreaded: schedule the chat sweeps, report connections in health Two background loops, both under the existing panic supervision, and one new health field. Retention runs every 15 minutes. The interval controls only how long an expired message lingers, since expiry is stamped on the row at insert and nothing is kept alive by a late sweep. Both statements run every round even when the first finds nothing: skipping the cap trim when the expiry sweep is empty would skip it in exactly the case that matters, because a room busy enough to be over its cap is usually one whose messages are all too new to have expired. There is a test for that shape. Rate-limit bucket eviction is a separate loop on the crate's own interval, not one chosen here. Evicting a bucket resets its budget, so sweeping faster than a bucket refills would turn eviction into a way to skip the queue; Chat::sweep_interval is what knows that. The buckets are keyed by (user, room), which is unbounded and attacker-influenced, so this is what keeps them off the 512M cap rather than housekeeping. One sweep round is split out as sweep_chat_once so it can be tested without waiting an interval, and it is convergent: a second round immediately after the first finds nothing. /api/health gains chat_connections. The design note asks whether it belongs in the PoM payload and it does: it is the number that predicts the cgroup cap, since each listener holds a task, a buffer and a broadcast receiver. Additive, and like tls_trust_anchors it does not move status or the HTTP code. No new env vars, so config_check_env_file has nothing to learn before a promote, and the deploy script derives its shared-crate rsync list from the manifests, so livechat needs no edit there either. 361 integration and 192 unit tests pass, clippy and fmt clean.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-08-08 19:54 UTC
Signed with PGP, not checked
Commit: e58daf7ef7e8ac5e0b9b196ad8a2a7018f007ee0
Parent: 0887db0
5 files changed, +210 insertions, -6 deletions
M multithreaded/src/main.rs +22
@@ -126,6 +126,24 @@
126 126 })
127 127 });
128 128
129 + // Chat retention: expire by age and trim each room to its message cap.
130 + // Unconditional. A deployment where no community has chat enabled runs two
131 + // indexed statements that match nothing, which is cheaper than the check.
132 + let chat_sweep_db = state.db.clone();
133 + let chat_sweep_task = supervise("chat-retention-sweep", move || {
134 + multithreaded::maintenance::continuously_sweep_chat(
135 + chat_sweep_db.clone(),
136 + multithreaded::maintenance::CHAT_SWEEP_INTERVAL,
137 + )
138 + });
139 +
140 + // Chat send-rate buckets. Keyed by (user, room), so the key space is
141 + // unbounded and attacker-influenced; this is what keeps it off the 512M cap.
142 + let chat_limiter = state.chat.clone();
143 + let chat_rate_task = supervise("chat-rate-bucket-sweep", move || {
144 + multithreaded::maintenance::continuously_sweep_chat_rate_limits(chat_limiter.clone())
145 + });
146 +
129 147 let session_layer = SessionManagerLayer::new(session_store)
130 148 .with_name("mt_session")
131 149 .with_same_site(SameSite::Lax)
@@ -217,6 +235,10 @@
217 235 task.abort();
218 236 let _ = task.await;
219 237 }
238 + for task in [chat_sweep_task, chat_rate_task] {
239 + task.abort();
240 + let _ = task.await;
241 + }
220 242 }
221 243
222 244 /// Spawn a never-returning background loop under panic supervision.
M multithreaded/src/maintenance.rs +91
@@ -76,3 +76,94 @@
76 76 }
77 77 Ok(purged)
78 78 }
79 +
80 + // Chat
81 +
82 + /// How often expired chat messages are swept.
83 + ///
84 + /// Sets how long an expired message lingers, which is the only thing the
85 + /// interval controls: expiry is stamped on the row at insert, so nothing is
86 + /// kept alive by a late sweep. Fifteen minutes keeps each delete small and
87 + /// bounds the lag well under the granularity anyone perceives in a window
88 + /// measured in days.
89 + pub const CHAT_SWEEP_INTERVAL: Duration = Duration::from_mins(15);
90 +
91 + /// Periodically enforce both halves of chat retention.
92 + ///
93 + /// Age and count are separate statements because they answer different
94 + /// questions: age expires a quiet room, and the count cap bounds a busy one
95 + /// that would reach its age limit holding far more than its owner allowed.
96 + /// Whichever bites first wins, which is what `livechat::Retention` documents.
97 + ///
98 + /// Both run every round even when the first finds nothing. Skipping the trim
99 + /// when the age sweep is empty would be wrong in exactly the case that matters:
100 + /// a room busy enough to be over its cap is usually one whose messages are all
101 + /// too new to have expired.
102 + ///
103 + /// Runs once at startup, then every `interval`. Cancel by aborting the task.
104 + pub async fn continuously_sweep_chat(db: PgPool, interval: Duration) {
105 + loop {
106 + sweep_chat_once(&db).await;
107 + tokio::time::sleep(interval).await;
108 + }
109 + }
110 +
111 + /// One round of both retention statements. Split out of the loop so it can be
112 + /// tested without waiting an interval.
113 + ///
114 + /// Returns `(expired, trimmed)`. Errors are logged rather than returned: a
115 + /// sweep is convergent and the next round retries, so a transient failure is
116 + /// not worth propagating into a supervisor restart.
117 + pub async fn sweep_chat_once(db: &PgPool) -> (u64, u64) {
118 + let expired = match mt_db::mutations::sweep_expired_chat_messages(db).await {
119 + Ok(n) => {
120 + if n > 0 {
121 + tracing::info!(expired = n, "chat: swept expired messages");
122 + }
123 + n
124 + }
125 + Err(e) => {
126 + tracing::error!(error = %e, "chat: expiry sweep failed");
127 + 0
128 + }
129 + };
130 +
131 + // Runs regardless of the result above: a failed expiry sweep is no reason
132 + // to let a room grow past its cap as well.
133 + let trimmed = match mt_db::mutations::trim_chat_rooms_to_cap(db).await {
134 + Ok(n) => {
135 + if n > 0 {
136 + tracing::info!(trimmed = n, "chat: trimmed rooms to their message cap");
137 + }
138 + n
139 + }
140 + Err(e) => {
141 + tracing::error!(error = %e, "chat: room cap trim failed");
142 + 0
143 + }
144 + };
145 +
146 + (expired, trimmed)
147 + }
148 +
149 + /// Periodically drop chat send-rate buckets nobody has touched recently.
150 + ///
151 + /// Not optional and not the same job as the retention sweep. Bucket state is
152 + /// keyed by (user, room), which is unbounded and attacker-influenced: anyone
153 + /// who can reach a room can mint a bucket, and both units run under a hard 512M
154 + /// cgroup cap where an OOM restarts the whole site rather than degrading chat.
155 + ///
156 + /// The crate picks the interval, and the choice is a correctness one rather
157 + /// than a tuning one: evicting a bucket resets its budget, so sweeping faster
158 + /// than a bucket can refill would turn eviction into a way to skip the queue.
159 + /// Hence `Chat::sweep_interval` rather than a number chosen here.
160 + pub async fn continuously_sweep_chat_rate_limits(chat: Arc<livechat::Chat>) {
161 + let interval = chat.sweep_interval();
162 + loop {
163 + tokio::time::sleep(interval).await;
164 + let dropped = chat.sweep(std::time::Instant::now());
165 + if dropped > 0 {
166 + tracing::debug!(dropped, "chat: evicted idle rate-limit buckets");
167 + }
168 + }
169 + }
M multithreaded/src/routes/mod.rs +18 -6
@@ -535,7 +535,11 @@
535 535
536 536 (
537 537 health_status(db_ok),
538 - Json(health_body(db_ok, crate::trust_store::anchors_ok())),
538 + Json(health_body(
539 + db_ok,
540 + crate::trust_store::anchors_ok(),
541 + state.chat.hub().connection_count(),
542 + )),
539 543 )
540 544 }
541 545
@@ -555,7 +559,7 @@
555 559 /// test in this module can exercise it directly. PoM polls this endpoint
556 560 /// and runs key-by-key assertions from `pom/deploy/pom-hetzner.toml`; the
557 561 /// guard test validates that every asserted path still resolves here.
558 - fn health_body(db_ok: bool, trust_anchors_ok: bool) -> serde_json::Value {
562 + fn health_body(db_ok: bool, trust_anchors_ok: bool, chat_connections: usize) -> serde_json::Value {
559 563 let status = if db_ok { "operational" } else { "degraded" };
560 564 serde_json::json!({
561 565 "status": status,
@@ -572,6 +576,14 @@
572 576 // the load-balancer check would turn a login outage into a total one.
573 577 // PoM asserts this field, which is what gets it monitored.
574 578 "tls_trust_anchors": trust_anchors_ok,
579 + // Live chat listeners across every room in this process. Exposed
580 + // because it is the number that predicts the 512M cgroup cap: each
581 + // listener holds a task, a buffer and a broadcast receiver, and an OOM
582 + // restarts the whole site rather than degrading chat. Like
583 + // `tls_trust_anchors` it deliberately does not move `status` or the
584 + // HTTP code; it is a trend for an operator to watch, and the hub
585 + // already refuses connections past its own cap.
586 + "chat_connections": chat_connections,
575 587 })
576 588 }
577 589
@@ -617,7 +629,7 @@
617 629 #[test]
618 630 #[ignore = "cross-repo: run by sweep's pom-contract check, which materializes pom"]
619 631 fn pom_hetzner_health_expectations_resolve() {
620 - let body = health_body(true, true);
632 + let body = health_body(true, true, 0);
621 633 pom_contract::assert_health_expectations_resolve(
622 634 "../pom/deploy/pom-hetzner.toml",
623 635 "mt",
@@ -631,7 +643,7 @@
631 643 /// than in PoM's exact-match `json_fields`.
632 644 #[test]
633 645 fn health_body_carries_version_and_git_sha_keys() {
634 - let body = health_body(true, true);
646 + let body = health_body(true, true, 0);
635 647 assert_eq!(body["version"], env!("CARGO_PKG_VERSION"));
636 648 assert!(
637 649 body.get("git_sha").is_some(),
@@ -645,9 +657,9 @@
645 657 /// leave it; degrading the whole target would hide that distinction.
646 658 #[test]
647 659 fn health_body_reports_trust_anchors_without_moving_status() {
648 - assert_eq!(health_body(true, true)["tls_trust_anchors"], true);
660 + assert_eq!(health_body(true, true, 0)["tls_trust_anchors"], true);
649 661
650 - let body = health_body(true, false);
662 + let body = health_body(true, false, 0);
651 663 assert_eq!(body["tls_trust_anchors"], false);
652 664 assert_eq!(body["status"], "operational");
653 665 assert_eq!(health_status(true), StatusCode::OK);
M multithreaded/tests/workflows/chat_routes.rs +17
@@ -492,3 +492,20 @@
492 492 // would hang rather than fail. Frame format and backlog-then-live ordering are
493 493 // covered by the crate's own tests (`livechat::sse`, `livechat::stream`); the
494 494 // end-to-end path belongs with the client island.
495 +
496 + // Health
497 +
498 + #[sqlx::test]
499 + async fn the_health_payload_reports_the_chat_connection_count(_pool: sqlx::PgPool) {
500 + // It is the number that predicts the 512M cgroup cap, so PoM can watch it.
501 + let mut h = TestHarness::new().await;
502 +
503 + let resp = h.client.get("/api/health").await;
504 + assert_eq!(resp.status, StatusCode::OK);
505 +
506 + let body: serde_json::Value = serde_json::from_str(&resp.text).expect("json");
507 + assert_eq!(
508 + body["chat_connections"], 0,
509 + "present and zero with nobody connected"
510 + );
511 + }
M multithreaded/tests/workflows/chat_storage.rs +62
@@ -442,3 +442,65 @@
442 442 assert!(result.is_err(), "{column} = {value} must be refused");
443 443 }
444 444 }
445 +
446 + // The scheduled sweep
447 +
448 + #[sqlx::test]
449 + async fn one_sweep_round_enforces_both_halves_of_retention(_pool: sqlx::PgPool) {
450 + // The two statements answer different questions and both must run every
451 + // round. A room busy enough to be over its cap is usually one whose
452 + // messages are all too new to have expired, so gating the trim on the
453 + // expiry sweep finding something would skip it exactly when it is needed.
454 + let mut h = TestHarness::new().await;
455 + let r = room(&mut h).await;
456 +
457 + sqlx::query("UPDATE communities SET chat_max_messages = 2 WHERE id = $1")
458 + .bind(r.id)
459 + .execute(&h.db)
460 + .await
461 + .unwrap();
462 +
463 + // Five fresh messages: nothing is expired, and three are over the cap.
464 + for i in 0..5 {
465 + say(&h, &r, r.alice, &format!("m{i}")).await;
466 + }
467 +
468 + let (expired, trimmed) = multithreaded::maintenance::sweep_chat_once(&h.db).await;
469 +
470 + assert_eq!(expired, 0, "nothing was old enough to expire");
471 + assert_eq!(trimmed, 3, "the cap still bit");
472 + assert_eq!(all(&h, &r).await, vec!["m3", "m4"]);
473 + }
474 +
475 + #[sqlx::test]
476 + async fn a_sweep_round_over_an_empty_table_is_a_no_op(_pool: sqlx::PgPool) {
477 + let h = TestHarness::new().await;
478 + assert_eq!(
479 + multithreaded::maintenance::sweep_chat_once(&h.db).await,
480 + (0, 0)
481 + );
482 + }
483 +
484 + #[sqlx::test]
485 + async fn the_sweep_is_convergent(_pool: sqlx::PgPool) {
486 + // Steady-state work is zero: a second round immediately after the first
487 + // must find nothing, or the sweep would churn the same rows every interval.
488 + let mut h = TestHarness::new().await;
489 + let r = room(&mut h).await;
490 +
491 + sqlx::query("UPDATE communities SET chat_max_messages = 2 WHERE id = $1")
492 + .bind(r.id)
493 + .execute(&h.db)
494 + .await
495 + .unwrap();
496 + for i in 0..5 {
497 + say(&h, &r, r.alice, &format!("m{i}")).await;
498 + }
499 +
500 + multithreaded::maintenance::sweep_chat_once(&h.db).await;
501 + assert_eq!(
502 + multithreaded::maintenance::sweep_chat_once(&h.db).await,
503 + (0, 0),
504 + "the second round must find nothing left to do"
505 + );
506 + }