Skip to main content

max / makenotwork

mt: stream image serves, rate-limit, segregate pools, bound governor (ultra-fuzz S1/Mi1/M-Pf1-3) S1: serve_image_handler streamed the whole object into a per-request Vec<u8> on an un-rate-limited, un-CDN'd path. Stream it straight from S3 to the response body (ByteStream -> tokio AsyncBufRead -> ReaderStream -> Body::from_stream), content-type from the stored row. The optional 2-PK-lookup query collapse is left as a noted micro-opt. Mi1: /uploads/{id} moved into its own per-IP rate-limited group (generous burst 60 for image-heavy pages) — it was an unauthenticated S3-egress proxy with no limiter. M-Pf1: the admin dashboard fetched LIMIT 500 silently; now CAP+1 with a "showing the first N; more exist" notice. M-Pf2: the Postgres session store gets its own 6-conn pool so session I/O and the expiry sweep stop contending with handler queries; handler pool raised to 32. M-Pf3: a 5-minute task calls retain_recent() on every rate limiter so the per-IP keyspace can't grow unbounded. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 06:29 UTC
Commit: 2290bc62ca39aebf906a9883cccbe04b8236d33f
Parent: ae9d34a
12 files changed, +163 insertions, -64 deletions
D multithreaded/.sqlx/query-6f586c26815fbe1b6880ff5bf18e2061248e4182d371a1a0b005dcacfc7543a8.json -44
@@ -1,44 +0,0 @@
1 - {
2 - "db_name": "PostgreSQL",
3 - "query": "SELECT id, name, slug,\n suspended_at AS \"suspended_at: chrono::DateTime<chrono::Utc>\",\n suspension_reason\n FROM communities ORDER BY name LIMIT 500",
4 - "describe": {
5 - "columns": [
6 - {
7 - "ordinal": 0,
8 - "name": "id",
9 - "type_info": "Uuid"
10 - },
11 - {
12 - "ordinal": 1,
13 - "name": "name",
14 - "type_info": "Text"
15 - },
16 - {
17 - "ordinal": 2,
18 - "name": "slug",
19 - "type_info": "Text"
20 - },
21 - {
22 - "ordinal": 3,
23 - "name": "suspended_at: chrono::DateTime<chrono::Utc>",
24 - "type_info": "Timestamptz"
25 - },
26 - {
27 - "ordinal": 4,
28 - "name": "suspension_reason",
29 - "type_info": "Text"
30 - }
31 - ],
32 - "parameters": {
33 - "Left": []
34 - },
35 - "nullable": [
36 - false,
37 - false,
38 - false,
39 - true,
40 - true
41 - ]
42 - },
43 - "hash": "6f586c26815fbe1b6880ff5bf18e2061248e4182d371a1a0b005dcacfc7543a8"
44 - }
A multithreaded/.sqlx/query-c79b4fda201c812450aa80b328b01192c4684029920b6ea2ab0b38a776ec1281.json +46
@@ -0,0 +1,46 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "SELECT id, name, slug,\n suspended_at AS \"suspended_at: chrono::DateTime<chrono::Utc>\",\n suspension_reason\n FROM communities ORDER BY name LIMIT $1",
4 + "describe": {
5 + "columns": [
6 + {
7 + "ordinal": 0,
8 + "name": "id",
9 + "type_info": "Uuid"
10 + },
11 + {
12 + "ordinal": 1,
13 + "name": "name",
14 + "type_info": "Text"
15 + },
16 + {
17 + "ordinal": 2,
18 + "name": "slug",
19 + "type_info": "Text"
20 + },
21 + {
22 + "ordinal": 3,
23 + "name": "suspended_at: chrono::DateTime<chrono::Utc>",
24 + "type_info": "Timestamptz"
25 + },
26 + {
27 + "ordinal": 4,
28 + "name": "suspension_reason",
29 + "type_info": "Text"
30 + }
31 + ],
32 + "parameters": {
33 + "Left": [
34 + "Int8"
35 + ]
36 + },
37 + "nullable": [
38 + false,
39 + false,
40 + false,
41 + true,
42 + true
43 + ]
44 + },
45 + "hash": "c79b4fda201c812450aa80b328b01192c4684029920b6ea2ab0b38a776ec1281"
46 + }
M multithreaded/Cargo.toml +2
@@ -14,6 +14,7 @@ license-file = "LICENSE"
14 14 [workspace.dependencies]
15 15 # Core
16 16 tokio = { version = "1", features = ["rt-multi-thread", "sync", "time", "macros", "signal"] }
17 + tokio-util = { version = "0.7", features = ["io"] }
17 18 thiserror = "2"
18 19 serde = { version = "1", features = ["derive"] }
19 20 serde_json = "1"
@@ -64,6 +65,7 @@ edition.workspace = true
64 65 mt-core = { workspace = true }
65 66 mt-db = { workspace = true }
66 67 tokio = { workspace = true }
68 + tokio-util = { workspace = true }
67 69 axum = { workspace = true }
68 70 tower = { workspace = true }
69 71 tower-http = { workspace = true }
M multithreaded/crates/mt-db/src/queries.rs +3 -1
@@ -1077,13 +1077,15 @@ pub struct AdminCommunityRow {
1077 1077 #[tracing::instrument(skip_all)]
1078 1078 pub async fn list_all_communities(
1079 1079 pool: &PgPool,
1080 + limit: i64,
1080 1081 ) -> Result<Vec<AdminCommunityRow>, sqlx::Error> {
1081 1082 sqlx::query_as!(
1082 1083 AdminCommunityRow,
1083 1084 r#"SELECT id, name, slug,
1084 1085 suspended_at AS "suspended_at: chrono::DateTime<chrono::Utc>",
1085 1086 suspension_reason
1086 - FROM communities ORDER BY name LIMIT 500"#,
1087 + FROM communities ORDER BY name LIMIT $1"#,
1088 + limit,
1087 1089 )
1088 1090 .fetch_all(pool)
1089 1091 .await
M multithreaded/src/main.rs +17 -7
@@ -19,12 +19,13 @@ async fn main() {
19 19 let database_url = std::env::var("DATABASE_URL")
20 20 .expect("DATABASE_URL must be set");
21 21
22 - // Explicit pool sizing. The sqlx default is 10 connections, and this pool
23 - // is shared with the tower-sessions PostgresStore (every authed request
24 - // does session I/O on top of its handler queries), so the default is tight.
25 - // Bound acquisition so a burst fails fast rather than hanging for 30s.
22 + // Explicit pool sizing for the handler pool. Some handlers fan out ~3
23 + // queries concurrently via `try_join!`, so 32 connections sustain ~10
24 + // concurrent such requests before acquisition queues. Session I/O and its
25 + // expiry sweep run on a *separate* pool (below), so they no longer contend
26 + // here (M-Pf2). Bound acquisition so a burst sheds fast rather than hanging.
26 27 let pool = PgPoolOptions::new()
27 - .max_connections(20)
28 + .max_connections(32)
28 29 .acquire_timeout(std::time::Duration::from_secs(10))
29 30 .connect(&database_url)
30 31 .await
@@ -78,8 +79,17 @@ async fn main() {
78 79 s3,
79 80 };
80 81
81 - // Session store backed by PostgreSQL
82 - let session_store = PostgresStore::new(pool);
82 + // Session store backed by PostgreSQL, on its own small pool. Every authed
83 + // request does session I/O and the hourly expiry sweep scans the session
84 + // table; keeping them off the handler pool means a handler-query burst can't
85 + // starve session reads (or vice versa) (M-Pf2).
86 + let session_pool = PgPoolOptions::new()
87 + .max_connections(6)
88 + .acquire_timeout(std::time::Duration::from_secs(10))
89 + .connect(&database_url)
90 + .await
91 + .expect("failed to connect session store pool");
92 + let session_store = PostgresStore::new(session_pool);
83 93 session_store.migrate().await.expect("failed to migrate session store");
84 94
85 95 // Both background loops run under panic supervision: a panic inside a sweep
M multithreaded/src/routes/admin.rs +10 -2
@@ -29,12 +29,19 @@ pub(super) async fn admin_dashboard(
29 29 ) -> Result<impl IntoResponse, Response> {
30 30 let csrf_token = Some(csrf::get_or_create_token(&session).await);
31 31
32 - let communities = mt_db::queries::list_all_communities(&state.db)
32 + // Cap the listing and surface overflow rather than silently truncating
33 + // (mirrors the moderation lists' CAP+1). Fetch CAP+1 so we can tell whether
34 + // more communities exist than we show.
35 + const ADMIN_COMMUNITY_CAP: usize = 500;
36 + let mut community_rows = mt_db::queries::list_all_communities(&state.db, ADMIN_COMMUNITY_CAP as i64 + 1)
33 37 .await
34 38 .map_err(|e| {
35 39 tracing::error!(error = ?e, "db error listing communities");
36 40 (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response()
37 - })?
41 + })?;
42 + let communities_truncated = community_rows.len() > ADMIN_COMMUNITY_CAP;
43 + community_rows.truncate(ADMIN_COMMUNITY_CAP);
44 + let communities = community_rows
38 45 .into_iter()
39 46 .map(|c| AdminCommunityViewRow {
40 47 id: c.id.to_string(),
@@ -74,6 +81,7 @@ pub(super) async fn admin_dashboard(
74 81 }),
75 82 mnw_base_url: state.config.mnw_base_url.clone(),
76 83 communities,
84 + communities_truncated,
77 85 users,
78 86 search_query,
79 87 })
M multithreaded/src/routes/mod.rs +56 -4
@@ -55,6 +55,14 @@ const MAX_UPLOAD_BODY_BYTES: usize = crate::storage::MAX_IMAGE_SIZE + 64 * 1024;
55 55 const AUTH_RATE_LIMIT_MS: u64 = 1000;
56 56 const AUTH_RATE_LIMIT_BURST: u32 = 10;
57 57
58 + /// Image serve (`/uploads/{id}`): generous, since a single page legitimately
59 + /// fans out one request per embedded `<img>`. Burst 60 covers an image-heavy
60 + /// page load, then refills ~20/sec — enough to bound an unauthenticated
61 + /// S3-egress proxy against a scraping/amplification flood (ultra-fuzz Mi1)
62 + /// without throttling normal browsing.
63 + const IMAGE_RATE_LIMIT_MS: u64 = 50;
64 + const IMAGE_RATE_LIMIT_BURST: u32 = 60;
65 +
58 66 /// Build the forum route tree.
59 67 pub fn forum_routes(state: AppState) -> Router {
60 68 let write_rate_limit = std::sync::Arc::new(
@@ -107,7 +115,7 @@ pub fn forum_routes(state: AppState) -> Router {
107 115 )
108 116 .route("/p/{slug}/uploads/{id}/remove", post(uploads::remove_image_handler))
109 117 .route_layer(GovernorLayer {
110 - config: write_rate_limit,
118 + config: write_rate_limit.clone(),
111 119 });
112 120
113 121 // Search — rate limited per IP (expensive full-text queries)
@@ -123,7 +131,7 @@ pub fn forum_routes(state: AppState) -> Router {
123 131 let search_routes = Router::new()
124 132 .route("/search", get(search::search_handler))
125 133 .route_layer(GovernorLayer {
126 - config: search_rate_limit,
134 + config: search_rate_limit.clone(),
127 135 });
128 136
129 137 // Auth endpoints — rate limited per IP (login/callback flood + refresh
@@ -144,9 +152,53 @@ pub fn forum_routes(state: AppState) -> Router {
144 152 .route("/auth/logout", post(auth::logout))
145 153 .route("/auth/refresh", post(auth::refresh))
146 154 .route_layer(GovernorLayer {
147 - config: auth_rate_limit,
155 + config: auth_rate_limit.clone(),
148 156 });
149 157
158 + // Image serve — per-IP rate limited. `/uploads/{id}` is an unauthenticated
159 + // S3-egress proxy (it streams bytes to any viewer who passes the community
160 + // access check), so a generous governor bounds scraping/amplification floods
161 + // (ultra-fuzz Mi1) while leaving image-heavy page loads unthrottled.
162 + let image_rate_limit = std::sync::Arc::new(
163 + GovernorConfigBuilder::default()
164 + .key_extractor(TrustedProxyKeyExtractor::new(state.config.trusted_proxies.clone()))
165 + .per_millisecond(IMAGE_RATE_LIMIT_MS)
166 + .burst_size(IMAGE_RATE_LIMIT_BURST)
167 + .finish()
168 + .expect("image rate limiter config"),
169 + );
170 +
171 + let image_routes = Router::new()
172 + .route("/uploads/{id}", get(uploads::serve_image_handler))
173 + .route_layer(GovernorLayer {
174 + config: image_rate_limit.clone(),
175 + });
176 +
177 + // Periodically evict idle per-IP buckets from every rate limiter so the
178 + // keyspace can't grow unbounded over the process lifetime (M-Pf3).
179 + // Trusted-proxy keying already bounds keys to real client IPs, but a
180 + // long-running server still accumulates one-off visitors; `retain_recent`
181 + // drops buckets with no recent activity. Spawned here because `forum_routes`
182 + // runs inside the tokio runtime at startup.
183 + {
184 + let limiters = [
185 + write_rate_limit.limiter().clone(),
186 + search_rate_limit.limiter().clone(),
187 + auth_rate_limit.limiter().clone(),
188 + image_rate_limit.limiter().clone(),
189 + ];
190 + tokio::spawn(async move {
191 + let mut interval = tokio::time::interval(std::time::Duration::from_secs(300));
192 + interval.tick().await; // consume the immediate first tick
193 + loop {
194 + interval.tick().await;
195 + for limiter in &limiters {
196 + limiter.retain_recent();
197 + }
198 + }
199 + });
200 + }
201 +
150 202 // GET routes + health — no rate limiting
151 203 let read_routes = Router::new()
152 204 .route("/", get(forum::forum_directory))
@@ -167,12 +219,12 @@ pub fn forum_routes(state: AppState) -> Router {
167 219 .route("/_admin", get(admin::admin_dashboard))
168 220 .route("/_admin/communities/{slug}", get(admin::admin_community_detail))
169 221 .route("/api/user/{user_id}/summary", get(forum::user_summary_api))
170 - .route("/uploads/{id}", get(uploads::serve_image_handler))
171 222 .route("/api/health", get(health));
172 223
173 224 read_routes
174 225 .merge(search_routes)
175 226 .merge(auth_routes)
227 + .merge(image_routes)
176 228 .merge(write_routes)
177 229 .fallback(not_found_handler)
178 230 .with_state(state)
M multithreaded/src/routes/uploads.rs +8 -5
@@ -1,7 +1,6 @@
1 1 //! Image upload and serving handlers.
2 2
3 3 use axum::{
4 - body::Body,
5 4 extract::{Multipart, Path},
6 5 http::{StatusCode, header},
7 6 response::{IntoResponse, Response},
@@ -172,8 +171,12 @@ pub(super) async fn serve_image_handler(
172 171 StatusCode::SERVICE_UNAVAILABLE.into_response()
173 172 })?;
174 173
175 - let (data, content_type) = s3.download(&image.s3_key).await.map_err(|e| {
176 - tracing::error!(error = %e, "S3 download failed");
174 + // Stream the object straight from S3 to the client instead of buffering the
175 + // whole image into a Vec<u8> per request (ultra-fuzz S1). The Content-Type
176 + // comes from the stored row (set from upload-time magic-byte validation), so
177 + // we don't need S3's metadata round-trip.
178 + let body = s3.download_stream(&image.s3_key).await.map_err(|e| {
179 + tracing::error!(error = %e, "S3 stream open failed");
177 180 StatusCode::INTERNAL_SERVER_ERROR.into_response()
178 181 })?;
179 182
@@ -185,11 +188,11 @@ pub(super) async fn serve_image_handler(
185 188 // validation, not a disposition flag, is what makes inline serving safe.
186 189 Ok(Response::builder()
187 190 .status(StatusCode::OK)
188 - .header(header::CONTENT_TYPE, content_type)
191 + .header(header::CONTENT_TYPE, image.content_type)
189 192 .header(header::CACHE_CONTROL, "private, max-age=86400, immutable")
190 193 .header(header::X_CONTENT_TYPE_OPTIONS, "nosniff")
191 194 .header(header::CONTENT_DISPOSITION, "inline")
192 - .body(Body::from(data))
195 + .body(body)
193 196 .unwrap())
194 197 }
195 198
M multithreaded/src/storage.rs +14
@@ -61,6 +61,20 @@ impl S3Storage {
61 61 self.inner.download(s3_key).await
62 62 }
63 63
64 + /// Open a streaming download from S3 as an HTTP response body, without
65 + /// buffering the object in memory.
66 + ///
67 + /// The image-serve handler streams the result to the client chunk-by-chunk
68 + /// instead of materialising a full `Vec<u8>` per request (ultra-fuzz S1). The
69 + /// aws `ByteStream` is adapted to a body via its tokio `AsyncBufRead` reader,
70 + /// keeping the aws/tokio-util plumbing contained here.
71 + #[tracing::instrument(skip_all)]
72 + pub async fn download_stream(&self, s3_key: &str) -> Result<axum::body::Body, String> {
73 + let stream = self.inner.download_stream(s3_key).await?;
74 + let reader = stream.into_async_read();
75 + Ok(axum::body::Body::from_stream(tokio_util::io::ReaderStream::new(reader)))
76 + }
77 +
64 78 /// Delete an object from S3.
65 79 #[tracing::instrument(skip_all)]
66 80 pub async fn delete(&self, s3_key: &str) -> Result<(), String> {
M multithreaded/src/templates/public.rs +3
@@ -645,6 +645,9 @@ pub struct AdminDashboardTemplate {
645 645 pub session_user: Option<TemplateSessionUser>,
646 646 pub mnw_base_url: std::sync::Arc<str>,
647 647 pub communities: Vec<AdminCommunityViewRow>,
648 + /// True when more communities exist than the page caps at, so the template
649 + /// can say so instead of silently dropping the tail.
650 + pub communities_truncated: bool,
648 651 pub users: Vec<AdminUserViewRow>,
649 652 pub search_query: String,
650 653 }
M multithreaded/templates/pages/admin.html +3
@@ -75,6 +75,9 @@
75 75
76 76 <div class="settings-section">
77 77 <h3>Communities</h3>
78 + {% if communities_truncated %}
79 + <div class="empty-state">Showing the first {{ communities.len() }} communities; more exist. Use search to find a specific one.</div>
80 + {% endif %}
78 81 {% if communities.is_empty() %}
79 82 <div class="empty-state">No communities.</div>
80 83 {% else %}
M multithreaded/tests/workflows/admin_queries.rs +1 -1
@@ -18,7 +18,7 @@ async fn test_list_all_communities() {
18 18 let _c2 = h.create_community("Beta Forum", "beta").await;
19 19 let _c3 = h.create_community("Gamma Forum", "gamma").await;
20 20
21 - let rows = mt_db::queries::list_all_communities(&h.db)
21 + let rows = mt_db::queries::list_all_communities(&h.db, 500)
22 22 .await
23 23 .unwrap();
24 24