Skip to main content

max / makenotwork

Multithreaded second sweep: Community/Storage/UX to A Take the A- axes from ultra-fuzz Run #1 up to A (Security stays A-, capped by cross-repo S13/HMAC items that are post-launch). Community: mod_remove_post_cascade soft-deletes the thread when its opening post is removed (atomic); wired into both mod-removal handlers. get_thread_with_breadcrumb now filters deleted_at so a soft-deleted thread 404s instead of staying reachable and repliable by direct link. Storage: migration 028 adds images.s3_purged_at; a background reconcile sweep purges orphaned S3 objects convergently and retries failed inline deletes. remove_image_handler marks purged on successful delete. UX: field_error returns 422 + X-Form-Field for content validations; mt.js renders a persistent inline error next to the named input (highlight/focus, input preserved) instead of a transient toast. Performance: keyset pagination and streaming uploads reviewed and deliberately deferred (net-negative at this scale). 359 tests green (117 unit + 242 integration), clippy clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-15 19:54 UTC
Commit: 873dc0667fd37e9d0265fc0498fc13f5ad7f7057
Parent: 9e7cbde
18 files changed, +641 insertions, -40 deletions
M multithreaded/crates/mt-db/src/mutations.rs +78
@@ -283,6 +283,69 @@ pub async fn mod_remove_post(
283 283 Ok(result.rows_affected() > 0)
284 284 }
285 285
286 + /// Outcome of [`mod_remove_post_cascade`].
287 + #[derive(Debug, Clone, Copy, PartialEq, Eq)]
288 + pub struct PostRemoval {
289 + /// The post was removed by this call (false if it was already removed).
290 + pub post_removed: bool,
291 + /// The post was the thread's opening post, so the whole thread was
292 + /// soft-deleted as well.
293 + pub thread_removed: bool,
294 + }
295 +
296 + /// Mod-remove a post and, if it is the thread's opening post, soft-delete the
297 + /// whole thread in the same transaction.
298 + ///
299 + /// The opening post is the earliest post in the thread (by `created_at`, `id`
300 + /// as tiebreak), counted regardless of removal state so the identity is stable.
301 + /// Removing an OP without this cascade would leave a headless, still-repliable
302 + /// thread; cascading deletes it so listings, the thread view, and the reply
303 + /// path all treat it as gone. Returns false flags when the post was already
304 + /// removed or was not the OP — callers can log a thread-deletion accordingly.
305 + #[tracing::instrument(skip_all)]
306 + pub async fn mod_remove_post_cascade(
307 + pool: &PgPool,
308 + post_id: Uuid,
309 + removed_by_id: Uuid,
310 + ) -> Result<PostRemoval, sqlx::Error> {
311 + let mut tx = pool.begin().await?;
312 +
313 + let post_removed = sqlx::query(
314 + "UPDATE posts SET removed_by = $2, removed_at = now()
315 + WHERE id = $1 AND removed_at IS NULL",
316 + )
317 + .bind(post_id)
318 + .bind(removed_by_id)
319 + .execute(&mut *tx)
320 + .await?
321 + .rows_affected()
322 + > 0;
323 +
324 + let mut thread_removed = false;
325 + if post_removed {
326 + // Soft-delete the thread only when this post is its opening post.
327 + thread_removed = sqlx::query(
328 + "UPDATE threads SET deleted_at = now()
329 + WHERE deleted_at IS NULL
330 + AND id = (SELECT thread_id FROM posts WHERE id = $1)
331 + AND $1 = (
332 + SELECT id FROM posts
333 + WHERE thread_id = (SELECT thread_id FROM posts WHERE id = $1)
334 + ORDER BY created_at ASC, id ASC
335 + LIMIT 1
336 + )",
337 + )
338 + .bind(post_id)
339 + .execute(&mut *tx)
340 + .await?
341 + .rows_affected()
342 + > 0;
343 + }
344 +
345 + tx.commit().await?;
346 + Ok(PostRemoval { post_removed, thread_removed })
347 + }
348 +
286 349 /// Atomically auto-hide a post if pending flag count meets the threshold.
287 350 /// Combines count check and removal in a single query to avoid race conditions.
288 351 /// Returns true if the post was actually removed.
@@ -1104,3 +1167,18 @@ pub async fn remove_image(
1104 1167 .await?;
1105 1168 Ok(())
1106 1169 }
1170 +
1171 + /// Mark images whose backing S3 object has been deleted, so the reconcile sweep
1172 + /// never revisits them. Called inline after a successful best-effort delete and
1173 + /// in batch by the background sweep.
1174 + #[tracing::instrument(skip_all)]
1175 + pub async fn mark_images_s3_purged(pool: &PgPool, image_ids: &[Uuid]) -> Result<(), sqlx::Error> {
1176 + if image_ids.is_empty() {
1177 + return Ok(());
1178 + }
1179 + sqlx::query("UPDATE images SET s3_purged_at = now() WHERE id = ANY($1)")
1180 + .bind(image_ids)
1181 + .execute(pool)
1182 + .await?;
1183 + Ok(())
1184 + }
M multithreaded/crates/mt-db/src/queries.rs +27 -1
@@ -422,7 +422,7 @@ pub async fn get_thread_with_breadcrumb(
422 422 FROM threads t
423 423 JOIN categories c ON c.id = t.category_id
424 424 JOIN communities co ON co.id = c.community_id
425 - WHERE t.id = $1",
425 + WHERE t.id = $1 AND t.deleted_at IS NULL",
426 426 )
427 427 .bind(thread_id)
428 428 .fetch_optional(pool)
@@ -1555,6 +1555,32 @@ pub async fn get_image(
1555 1555 .await
1556 1556 }
1557 1557
1558 + /// A removed image whose backing S3 object still needs to be purged.
1559 + #[derive(sqlx::FromRow)]
1560 + pub struct PendingPurgeImage {
1561 + pub id: Uuid,
1562 + pub s3_key: String,
1563 + }
1564 +
1565 + /// List removed images whose S3 object has not been purged yet, oldest first.
1566 + /// Drives the background reconcile sweep; `limit` bounds one batch (S3's
1567 + /// batch-delete cap is 1000).
1568 + #[tracing::instrument(skip_all)]
1569 + pub async fn list_images_pending_s3_purge(
1570 + pool: &PgPool,
1571 + limit: i64,
1572 + ) -> Result<Vec<PendingPurgeImage>, sqlx::Error> {
1573 + sqlx::query_as::<_, PendingPurgeImage>(
1574 + "SELECT id, s3_key FROM images
1575 + WHERE removed_at IS NOT NULL AND s3_purged_at IS NULL
1576 + ORDER BY removed_at ASC
1577 + LIMIT $1",
1578 + )
1579 + .bind(limit)
1580 + .fetch_all(pool)
1581 + .await
1582 + }
1583 +
1558 1584 /// Get the maximum sort_order among categories in a community. Returns -1 if no categories exist.
1559 1585 #[tracing::instrument(skip_all)]
1560 1586 pub async fn get_max_category_order(
A multithreaded/migrations/028_image_s3_purge.sql +11
@@ -0,0 +1,11 @@
1 + -- Track which removed images have had their backing S3 object deleted.
2 + -- `remove_image` sets removed_at but the object delete is best-effort; this
3 + -- column lets a background sweep find removed images whose object still needs
4 + -- purging (the pre-existing backlog and any inline-delete failures) and retry
5 + -- them convergently — once purged, an image is never revisited.
6 + ALTER TABLE images ADD COLUMN IF NOT EXISTS s3_purged_at TIMESTAMPTZ;
7 +
8 + -- The sweep scans for removed-but-not-purged images; keep that lookup cheap.
9 + CREATE INDEX IF NOT EXISTS idx_images_pending_purge
10 + ON images(removed_at)
11 + WHERE removed_at IS NOT NULL AND s3_purged_at IS NULL;
M multithreaded/src/lib.rs +1
@@ -5,6 +5,7 @@ pub mod config;
5 5 pub mod csrf;
6 6 pub mod internal_auth;
7 7 pub mod link_preview;
8 + pub mod maintenance;
8 9 pub mod routes;
9 10 pub mod seed;
10 11 pub mod storage;
M multithreaded/src/main.rs +14
@@ -87,6 +87,16 @@ async fn main() {
87 87 .continuously_delete_expired(tokio::time::Duration::from_secs(3600)),
88 88 );
89 89
90 + // Reconcile sweep: purge S3 objects for removed images (backlog + retries).
91 + // Only meaningful when S3 is configured.
92 + let purge_task = state.s3.as_ref().map(|s3| {
93 + tokio::task::spawn(multithreaded::maintenance::continuously_purge_removed_images(
94 + state.db.clone(),
95 + s3.clone(),
96 + tokio::time::Duration::from_secs(6 * 3600),
97 + ))
98 + });
99 +
90 100 let session_layer = SessionManagerLayer::new(session_store)
91 101 .with_name("mt_session")
92 102 .with_same_site(SameSite::Lax)
@@ -146,6 +156,10 @@ async fn main() {
146 156
147 157 deletion_task.abort();
148 158 let _ = deletion_task.await;
159 + if let Some(task) = purge_task {
160 + task.abort();
161 + let _ = task.await;
162 + }
149 163 }
150 164
151 165 async fn shutdown_signal() {
A multithreaded/src/maintenance.rs +78
@@ -0,0 +1,78 @@
1 + //! Background maintenance tasks.
2 +
3 + use std::collections::HashSet;
4 + use std::sync::Arc;
5 + use std::time::Duration;
6 +
7 + use sqlx::PgPool;
8 +
9 + use crate::storage::S3Storage;
10 +
11 + /// One batch of removed-image objects to delete per round. S3's batch-delete
12 + /// caps at 1000; staying well under keeps each request small.
13 + const PURGE_BATCH: i64 = 500;
14 +
15 + /// Periodically delete the S3 objects backing removed images whose object has
16 + /// not been purged yet.
17 + ///
18 + /// Closes two gaps left by the inline best-effort delete in
19 + /// `remove_image_handler`: images removed before purge-tracking existed (the
20 + /// backlog) and removals whose inline delete failed transiently. The sweep is
21 + /// convergent — every successfully deleted object's image is marked
22 + /// `s3_purged_at` and never revisited, so steady-state work is zero.
23 + ///
24 + /// Runs once at startup, then every `interval`. Cancel by aborting the task.
25 + pub async fn continuously_purge_removed_images(db: PgPool, s3: Arc<S3Storage>, interval: Duration) {
26 + loop {
27 + match purge_removed_image_objects(&db, &s3).await {
28 + Ok(0) => {}
29 + Ok(n) => tracing::info!(purged = n, "reconcile: purged orphaned image objects"),
30 + Err(e) => tracing::error!(error = %e, "reconcile: image purge sweep failed"),
31 + }
32 + tokio::time::sleep(interval).await;
33 + }
34 + }
35 +
36 + /// Delete the S3 objects for all removed-but-unpurged images, in batches.
37 + /// Returns the number of image objects purged. A batch that fails to make any
38 + /// progress (every key errored) stops the sweep so it retries next interval
39 + /// rather than spinning on the same failures.
40 + async fn purge_removed_image_objects(db: &PgPool, s3: &S3Storage) -> Result<usize, String> {
41 + let mut purged = 0usize;
42 + loop {
43 + let pending = mt_db::queries::list_images_pending_s3_purge(db, PURGE_BATCH)
44 + .await
45 + .map_err(|e| format!("db error listing images to purge: {e}"))?;
46 + if pending.is_empty() {
47 + break;
48 + }
49 +
50 + let keys: Vec<String> = pending.iter().map(|p| p.s3_key.clone()).collect();
51 + let failures = s3.delete_objects(&keys).await?;
52 + let failed: HashSet<&str> = failures.iter().map(|(k, _)| k.as_str()).collect();
53 + for (key, msg) in &failures {
54 + tracing::warn!(s3_key = %key, error = %msg, "reconcile: failed to delete image object");
55 + }
56 +
57 + let purged_ids: Vec<uuid::Uuid> = pending
58 + .iter()
59 + .filter(|p| !failed.contains(p.s3_key.as_str()))
60 + .map(|p| p.id)
61 + .collect();
62 +
63 + if purged_ids.is_empty() {
64 + // No progress this round — leave the rest for the next interval.
65 + break;
66 + }
67 +
68 + purged += purged_ids.len();
69 + mt_db::mutations::mark_images_s3_purged(db, &purged_ids)
70 + .await
71 + .map_err(|e| format!("db error marking images purged: {e}"))?;
72 +
73 + if pending.len() < PURGE_BATCH as usize {
74 + break;
75 + }
76 + }
77 + Ok(purged)
78 + }
M multithreaded/src/routes/account.rs +4 -5
@@ -18,7 +18,7 @@ use crate::templates::*;
18 18 use crate::AppState;
19 19
20 20 use super::{
21 - render_markdown, render_markdown_plus, template_user, SignatureForm,
21 + field_error, render_markdown, render_markdown_plus, template_user, SignatureForm,
22 22 };
23 23
24 24 const SIGNATURE_MAX: usize = 1024;
@@ -97,11 +97,10 @@ pub(super) async fn update_signature_handler(
97 97 return Ok(Redirect::to("/account"));
98 98 }
99 99 if trimmed.chars().count() > SIGNATURE_MAX {
100 - return Err((
101 - StatusCode::UNPROCESSABLE_ENTITY,
100 + return Err(field_error(
101 + "signature",
102 102 format!("Signature must be at most {SIGNATURE_MAX} characters."),
103 - )
104 - .into_response());
103 + ));
105 104 }
106 105
107 106 // Render with the same plus-aware paths as posts: creators get image
M multithreaded/src/routes/flagging.rs +21 -9
@@ -14,7 +14,8 @@ use crate::AppState;
14 14 use mt_core::types::ModAction;
15 15
16 16 use super::{
17 - check_community_access, get_community, log_mod_action, parse_uuid, require_mod_or_owner,
17 + check_community_access, field_error, get_community, log_mod_action, parse_uuid,
18 + require_mod_or_owner,
18 19 };
19 20
20 21 #[derive(Deserialize)]
@@ -38,7 +39,7 @@ pub(super) async fn flag_post_handler(
38 39
39 40 // Validate reason
40 41 if !matches!(form.reason.as_str(), "spam" | "rule_breaking" | "off_topic") {
41 - return Err((StatusCode::UNPROCESSABLE_ENTITY, "Invalid flag reason.").into_response());
42 + return Err(field_error("reason", "Invalid flag reason."));
42 43 }
43 44
44 45 // Fetch post to check ownership and existence
@@ -70,7 +71,7 @@ pub(super) async fn flag_post_handler(
70 71 if let Some(d) = detail
71 72 && d.len() > 1024
72 73 {
73 - return Err((StatusCode::UNPROCESSABLE_ENTITY, "Flag detail too long (max 1024 bytes).").into_response());
74 + return Err(field_error("detail", "Flag detail too long (max 1024 bytes)."));
74 75 }
75 76
76 77 mt_db::mutations::insert_flag(&state.db, post_id, user.user_id, &form.reason, detail)
@@ -153,9 +154,9 @@ pub(super) async fn remove_flagged_post_handler(
153 154 let (community, _role) = require_mod_or_owner(&state, &slug, &user).await?;
154 155 let flag_id = parse_uuid(&flag_id_str)?;
155 156
156 - // Get the flag to find the post_id — scoped to this community
157 - let flag_row: Option<(uuid::Uuid, uuid::Uuid)> = sqlx::query_as(
158 - "SELECT pf.post_id, p.author_id
157 + // Get the flag to find the post_id and its thread — scoped to this community
158 + let flag_row: Option<(uuid::Uuid, uuid::Uuid, uuid::Uuid)> = sqlx::query_as(
159 + "SELECT pf.post_id, p.author_id, t.id
159 160 FROM post_flags pf
160 161 JOIN posts p ON p.id = pf.post_id
161 162 JOIN threads t ON t.id = p.thread_id
@@ -171,11 +172,12 @@ pub(super) async fn remove_flagged_post_handler(
171 172 (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response()
172 173 })?;
173 174
174 - let (post_id, author_id) = flag_row
175 + let (post_id, author_id, thread_id) = flag_row
175 176 .ok_or_else(|| (StatusCode::NOT_FOUND, "Not found").into_response())?;
176 177
177 - // Mod-remove the post (idempotent — returns false if already removed)
178 - let _ = mt_db::mutations::mod_remove_post(&state.db, post_id, user.user_id)
178 + // Mod-remove the post (idempotent); if it is the OP, the whole thread is
179 + // soft-deleted in the same transaction.
180 + let removal = mt_db::mutations::mod_remove_post_cascade(&state.db, post_id, user.user_id)
179 181 .await
180 182 .map_err(|e| {
181 183 tracing::error!(error = ?e, "db error removing flagged post");
@@ -195,6 +197,16 @@ pub(super) async fn remove_flagged_post_handler(
195 197 ModAction::RemovePostViaFlag, Some(author_id), Some(post_id), None,
196 198 ).await;
197 199
200 + if removal.thread_removed {
201 + log_mod_action(
202 + &state.db, Some(community.id), user.user_id,
203 + ModAction::DeleteThread, Some(author_id), Some(thread_id), None,
204 + ).await;
205 + return Ok(Redirect::to(&format!(
206 + "/p/{slug}/moderation?toast=Thread+removed"
207 + )));
208 + }
209 +
198 210 Ok(Redirect::to(&format!(
199 211 "/p/{slug}/moderation?toast=Post+removed"
200 212 )))
M multithreaded/src/routes/helpers.rs +28 -11
@@ -183,30 +183,47 @@ pub(crate) fn template_user(
183 183 }
184 184 }
185 185
186 + /// Build a 422 validation response tagged with the offending form field.
187 + ///
188 + /// `field` matches the input's `name` attribute. The `X-Form-Field` header lets
189 + /// `mt.js` render the message as a persistent inline error next to that input
190 + /// (and focus it) instead of a transient toast — the A-grade form-failure UX.
191 + /// Submissions never navigate away on 422, so the user's input is preserved in
192 + /// the live DOM; only the error needs to come back.
193 + #[allow(clippy::result_large_err)]
194 + pub(crate) fn field_error(field: &'static str, message: impl Into<String>) -> Response {
195 + (
196 + StatusCode::UNPROCESSABLE_ENTITY,
197 + [("X-Form-Field", field)],
198 + message.into(),
199 + )
200 + .into_response()
201 + }
202 +
186 203 /// Validate a title field (1-256 chars).
187 204 #[allow(clippy::result_large_err)]
188 205 pub(crate) fn validate_title(text: &str) -> Result<&str, Response> {
189 206 let t = text.trim();
190 207 if t.is_empty() || t.len() > 256 {
191 - return Err((
192 - StatusCode::UNPROCESSABLE_ENTITY,
208 + return Err(field_error(
209 + "title",
193 210 "Title must be between 1 and 256 characters.",
194 - )
195 - .into_response());
211 + ));
196 212 }
197 213 Ok(t)
198 214 }
199 215
200 - /// Validate a body/content field (1 to max chars).
216 + /// Validate a body/content field (1 to max chars). `label` names the field in
217 + /// the message ("Body", "Footnote"); the inline error attaches to the `body`
218 + /// input, which every content textarea in the app uses as its `name`.
201 219 #[allow(clippy::result_large_err)]
202 - pub(crate) fn validate_body<'a>(text: &'a str, max: usize, field: &str) -> Result<&'a str, Response> {
220 + pub(crate) fn validate_body<'a>(text: &'a str, max: usize, label: &str) -> Result<&'a str, Response> {
203 221 let t = text.trim();
204 222 if t.is_empty() || t.len() > max {
205 - return Err((
206 - StatusCode::UNPROCESSABLE_ENTITY,
207 - format!("{field} must be between 1 and {max} characters."),
208 - )
209 - .into_response());
223 + return Err(field_error(
224 + "body",
225 + format!("{label} must be between 1 and {max} characters."),
226 + ));
210 227 }
211 228 Ok(t)
212 229 }
M multithreaded/src/routes/moderation.rs +19 -5
@@ -16,8 +16,9 @@ use crate::AppState;
16 16 use mt_core::types::{BanType, ModAction};
17 17
18 18 use super::{
19 - get_role, get_thread, get_user_by_username, is_mod_or_owner, is_owner, log_mod_action,
20 - parse_duration, parse_uuid, require_mod_or_owner, template_user, BanForm, PageQuery, UnbanForm,
19 + field_error, get_role, get_thread, get_user_by_username, is_mod_or_owner, is_owner,
20 + log_mod_action, parse_duration, parse_uuid, require_mod_or_owner, template_user, BanForm,
21 + PageQuery, UnbanForm,
21 22 };
22 23
23 24 #[tracing::instrument(skip_all)]
@@ -121,7 +122,7 @@ pub(super) async fn mod_remove_post_handler(
121 122 return Err(StatusCode::FORBIDDEN.into_response());
122 123 }
123 124
124 - let _ = mt_db::mutations::mod_remove_post(&state.db, post_id, user.user_id)
125 + let removal = mt_db::mutations::mod_remove_post_cascade(&state.db, post_id, user.user_id)
125 126 .await
126 127 .map_err(|e| {
127 128 tracing::error!(error = ?e, "db error removing post");
@@ -133,6 +134,19 @@ pub(super) async fn mod_remove_post_handler(
133 134 ModAction::RemovePost, Some(post_data.author_id), Some(post_id), None,
134 135 ).await;
135 136
137 + // Removing the opening post cascades to soft-deleting the whole thread; the
138 + // thread page now 404s, so send the mod back to the category listing.
139 + if removal.thread_removed {
140 + let thread_id = parse_uuid(&thread_id_str)?;
141 + log_mod_action(
142 + &state.db, Some(post_data.community_id), user.user_id,
143 + ModAction::DeleteThread, Some(post_data.author_id), Some(thread_id), None,
144 + ).await;
145 + return Ok(Redirect::to(&format!(
146 + "/p/{slug}/{category_slug}?toast=Thread+removed"
147 + )));
148 + }
149 +
136 150 Ok(Redirect::to(&format!(
137 151 "/p/{slug}/{category_slug}/{thread_id_str}?toast=Post+removed"
138 152 )))
@@ -250,7 +264,7 @@ pub(super) async fn ban_user_handler(
250 264 if let Some(r) = reason
251 265 && r.len() > 1024
252 266 {
253 - return Err((StatusCode::UNPROCESSABLE_ENTITY, "Reason too long (max 1024 bytes).").into_response());
267 + return Err(field_error("reason", "Reason too long (max 1024 bytes)."));
254 268 }
255 269
256 270 mt_db::mutations::create_community_ban(
@@ -335,7 +349,7 @@ pub(super) async fn mute_user_handler(
335 349 if let Some(r) = reason
336 350 && r.len() > 1024
337 351 {
338 - return Err((StatusCode::UNPROCESSABLE_ENTITY, "Reason too long (max 1024 bytes).").into_response());
352 + return Err(field_error("reason", "Reason too long (max 1024 bytes)."));
339 353 }
340 354
341 355 mt_db::mutations::create_community_ban(
M multithreaded/src/routes/uploads.rs +13 -5
@@ -216,11 +216,19 @@ pub(super) async fn remove_image_handler(
216 216
217 217 // Delete the backing S3 object so removed images don't accumulate in the
218 218 // bucket forever. Best-effort: the DB row is already marked removed (serve
219 - // returns 410), so a transient S3 failure is logged, not surfaced.
220 - if let Some(s3) = state.s3.as_ref()
221 - && let Err(e) = s3.delete(&image.s3_key).await
222 - {
223 - tracing::warn!(error = %e, s3_key = %image.s3_key, "failed to delete removed image from S3");
219 + // returns 410). On success, record s3_purged_at so the reconcile sweep skips
220 + // it; on failure, leave it unmarked so the sweep retries it later.
221 + if let Some(s3) = state.s3.as_ref() {
222 + match s3.delete(&image.s3_key).await {
223 + Ok(()) => {
224 + if let Err(e) = mt_db::mutations::mark_images_s3_purged(&state.db, &[image_id]).await {
225 + tracing::warn!(error = ?e, "failed to mark image S3-purged (sweep will retry)");
226 + }
227 + }
228 + Err(e) => {
229 + tracing::warn!(error = %e, s3_key = %image.s3_key, "failed to delete removed image from S3 (sweep will retry)");
230 + }
231 + }
224 232 }
225 233
226 234 // Log mod action
M multithreaded/src/storage.rs +8
@@ -58,6 +58,14 @@ impl S3Storage {
58 58 pub async fn delete(&self, s3_key: &str) -> Result<(), String> {
59 59 self.inner.delete(s3_key).await
60 60 }
61 +
62 + /// Batch-delete objects in a single request (S3 allows up to 1000 keys).
63 + /// Returns the keys that failed to delete, paired with the error message;
64 + /// an empty vec means every key was deleted (or was already absent).
65 + #[tracing::instrument(skip_all)]
66 + pub async fn delete_objects(&self, s3_keys: &[String]) -> Result<Vec<(String, String)>, String> {
67 + self.inner.delete_objects(s3_keys).await
68 + }
61 69 }
62 70
63 71 /// Generate an S3 key for a forum image.
M multithreaded/static/mt.js +54 -4
@@ -40,6 +40,49 @@ document.body.addEventListener('showToast', function(evt) {
40 40 showToast(evt.detail.message || 'Action completed', evt.detail.type || 'info');
41 41 });
42 42
43 + /* ===========================================
44 + INLINE FORM ERRORS
45 +
46 + A failed submit (422) keeps the user on the page with their input intact and
47 + shows a persistent error attached to the form — and, when the handler names
48 + the offending field via X-Form-Field, highlights and focuses that input.
49 + This replaces the transient error toast for validation failures.
50 + =========================================== */
51 +
52 + function clearFormError(form) {
53 + var prev = form.querySelector('.form-error');
54 + if (prev) prev.remove();
55 + form.querySelectorAll('[aria-invalid="true"]').forEach(function(el) {
56 + el.removeAttribute('aria-invalid');
57 + });
58 + }
59 +
60 + function showFormError(form, message, field) {
61 + clearFormError(form);
62 + var err = document.createElement('div');
63 + err.className = 'form-error';
64 + err.setAttribute('role', 'alert');
65 + err.textContent = message;
66 + form.insertBefore(err, form.firstChild);
67 +
68 + var focusTarget = null;
69 + if (field) {
70 + focusTarget = form.querySelector('[name="' + field + '"]');
71 + if (focusTarget) focusTarget.setAttribute('aria-invalid', 'true');
72 + }
73 + (focusTarget || err).scrollIntoView({ block: 'nearest', behavior: 'smooth' });
74 + if (focusTarget) focusTarget.focus();
75 + }
76 +
77 + // Clear a field's invalid state (and the form error) once the user edits it.
78 + document.addEventListener('input', function(e) {
79 + var field = e.target;
80 + if (field.getAttribute && field.getAttribute('aria-invalid') === 'true') {
81 + var form = field.closest('form');
82 + if (form) clearFormError(form);
83 + }
84 + });
85 +
43 86 document.body.addEventListener('htmx:responseError', function(evt) {
44 87 var container = document.getElementById('notifications');
45 88 if (!container) return;
@@ -200,6 +243,7 @@ document.addEventListener('submit', function(e) {
200 243 var token = document.querySelector('meta[name="csrf-token"]')?.content;
201 244 if (!token) return;
202 245 e.preventDefault();
246 + clearFormError(form);
203 247 fetch(form.action || window.location.href, {
204 248 method: 'POST',
205 249 headers: { 'X-CSRF-Token': token, 'Content-Type': 'application/x-www-form-urlencoded' },
@@ -213,11 +257,17 @@ document.addEventListener('submit', function(e) {
213 257 window.location.href = resp.url;
214 258 return;
215 259 }
216 - // Validation/other failure: keep the user on the page with their
217 - // input intact and surface the handler's message, rather than
218 - // navigating to the POST-only URL (which would GET a 404).
219 260 return resp.text().then(function(msg) {
220 - showToast(msg || 'Something went wrong. Please try again.', 'error');
261 + if (resp.status === 422) {
262 + // Validation failure: persistent inline error on the form,
263 + // input preserved, offending field highlighted/focused.
264 + showFormError(form, msg || 'Please check your input and try again.',
265 + resp.headers.get('X-Form-Field'));
266 + } else {
267 + // Auth/rate-limit/server errors aren't field problems —
268 + // a transient toast is the right surface.
269 + showToast(msg || 'Something went wrong. Please try again.', 'error');
270 + }
221 271 });
222 272 }).catch(function() {
223 273 showToast('Network error. Please try again.', 'error');
M multithreaded/static/style.css +17
@@ -341,6 +341,23 @@ textarea.drag-over {
341 341 background: var(--surface-muted);
342 342 }
343 343
344 + /* Inline form validation error — inserted by mt.js on a 422 submit, replacing
345 + the old error toast. Persists next to the form with the input preserved. */
346 + .form-error {
347 + background: var(--surface-muted);
348 + border-left: 3px solid var(--error);
349 + color: var(--error);
350 + padding: 0.6rem 0.75rem;
351 + margin-bottom: 0.75rem;
352 + font-size: 0.85rem;
353 + }
354 +
355 + input[aria-invalid="true"],
356 + textarea[aria-invalid="true"],
357 + select[aria-invalid="true"] {
358 + border: 2px solid var(--error);
359 + }
360 +
344 361 select {
345 362 cursor: pointer;
346 363 }
M multithreaded/tests/workflows/crud.rs +58
@@ -666,3 +666,61 @@ async fn quote_renders_with_attribution() {
666 666 "Thread page should contain quote attribution"
667 667 );
668 668 }
669 +
670 + // ============================================================================
671 + // Inline field-error contract — 422 + X-Form-Field so mt.js can render the
672 + // message next to the offending input instead of as a transient toast.
673 + // ============================================================================
674 +
675 + #[tokio::test]
676 + async fn empty_title_returns_inline_field_error() {
677 + let mut h = TestHarness::new().await;
678 + let user_id = h.login_as("fieldtitle").await;
679 + let comm_id = h.create_community("Test", "test").await;
680 + let _cat_id = h.create_category(comm_id, "General", "general").await;
681 + h.add_membership(user_id, comm_id, "member").await;
682 +
683 + h.client.get("/p/test/general/new").await;
684 + let resp = h
685 + .client
686 + .post_form("/p/test/general/new", "title=&body=Some+real+body")
687 + .await;
688 +
689 + assert_eq!(resp.status.as_u16(), 422, "Empty title should be rejected");
690 + assert_eq!(
691 + resp.headers.get("X-Form-Field").and_then(|v| v.to_str().ok()),
692 + Some("title"),
693 + "Validation error must name the title field for inline display"
694 + );
695 + assert!(
696 + resp.text.contains("Title"),
697 + "Error body should carry the human message, got: {}",
698 + resp.text
699 + );
700 + }
701 +
702 + #[tokio::test]
703 + async fn empty_reply_body_returns_inline_field_error() {
704 + let mut h = TestHarness::new().await;
705 + let user_id = h.login_as("fieldbody").await;
706 + let comm_id = h.create_community("Test", "test").await;
707 + let cat_id = h.create_category(comm_id, "General", "general").await;
708 + h.add_membership(user_id, comm_id, "member").await;
709 + let thread_id = h
710 + .create_thread_with_post(cat_id, user_id, "Thread", "OP body")
711 + .await;
712 +
713 + let thread_url = format!("/p/test/general/{}", thread_id);
714 + h.client.get(&thread_url).await;
715 + let resp = h
716 + .client
717 + .post_form(&format!("{}/reply", thread_url), "body=")
718 + .await;
719 +
720 + assert_eq!(resp.status.as_u16(), 422, "Empty reply body should be rejected");
721 + assert_eq!(
722 + resp.headers.get("X-Form-Field").and_then(|v| v.to_str().ok()),
723 + Some("body"),
724 + "Validation error must name the body field for inline display"
725 + );
726 + }
M multithreaded/tests/workflows/flagging.rs +46
@@ -397,3 +397,49 @@ async fn mod_remove_via_flag() {
397 397 .unwrap();
398 398 assert!(post.unwrap().0, "Post should be mod-removed");
399 399 }
400 +
401 + #[tokio::test]
402 + async fn remove_flagged_op_cascades_to_thread_delete() {
403 + let mut h = TestHarness::new().await;
404 + let author_id = h.login_as("flagcascadeauthor").await;
405 + let comm_id = h.create_community("Test", "test").await;
406 + let cat_id = h.create_category(comm_id, "General", "general").await;
407 + h.add_membership(author_id, comm_id, "member").await;
408 +
409 + let thread_id = h
410 + .create_thread_with_post(cat_id, author_id, "Flag Cascade", "Opening content")
411 + .await;
412 + let posts = mt_db::queries::list_posts_in_thread(&h.db, thread_id)
413 + .await
414 + .unwrap();
415 + let op_id = posts[0].id;
416 +
417 + // Flag the opening post.
418 + let flagger = h.login_as("flagcascadeflagger").await;
419 + h.add_membership(flagger, comm_id, "member").await;
420 + mt_db::mutations::insert_flag(&h.db, op_id, flagger, "spam", None)
421 + .await
422 + .unwrap();
423 +
424 + // Mod removes the flagged OP via the flag-remove endpoint.
425 + let mod_id = h.login_as("flagcascademod").await;
426 + h.add_membership(mod_id, comm_id, "moderator").await;
427 + let flags = mt_db::queries::list_pending_flags(&h.db, comm_id).await.unwrap();
428 + let flag_id = flags[0].flag_id;
429 + h.client.get("/p/test/moderation").await;
430 +
431 + let remove_url = format!("/p/test/moderation/flags/{}/remove", flag_id);
432 + let resp = h.client.post_form(&remove_url, "").await;
433 + assert!(resp.status.is_redirection(), "Expected redirect, got {}", resp.status);
434 +
435 + // The thread is soft-deleted and now 404s.
436 + let deleted: (bool,) = sqlx::query_as("SELECT deleted_at IS NOT NULL FROM threads WHERE id = $1")
437 + .bind(thread_id)
438 + .fetch_one(&h.db)
439 + .await
440 + .unwrap();
441 + assert!(deleted.0, "Removing the flagged OP should soft-delete the thread");
442 +
443 + let resp = h.client.get(&format!("/p/test/general/{}", thread_id)).await;
444 + assert_eq!(resp.status.as_u16(), 404, "Cascaded thread should 404");
445 + }
M multithreaded/tests/workflows/moderation.rs +102
@@ -472,3 +472,105 @@ async fn moderation_page_shows_bans_and_flags() {
472 472 "Moderation page should show pending flag reason"
473 473 );
474 474 }
475 +
476 + // ============================================================================
477 + // OP-removal cascade — removing the opening post soft-deletes the whole thread
478 + // ============================================================================
479 +
480 + #[tokio::test]
481 + async fn removing_op_cascades_to_thread_delete() {
482 + let mut h = TestHarness::new().await;
483 + let author_id = h.login_as("cascadeauthor").await;
484 + let comm_id = h.create_community("Test", "test").await;
485 + let cat_id = h.create_category(comm_id, "General", "general").await;
486 + h.add_membership(author_id, comm_id, "member").await;
487 +
488 + let thread_id = h
489 + .create_thread_with_post(cat_id, author_id, "Cascade Thread", "Opening content")
490 + .await;
491 +
492 + // Add a reply so the thread is more than just its OP.
493 + mt_db::mutations::create_post(&h.db, thread_id, author_id, "A reply", "<p>A reply</p>")
494 + .await
495 + .unwrap();
496 +
497 + let posts = mt_db::queries::list_posts_in_thread(&h.db, thread_id)
498 + .await
499 + .unwrap();
500 + let op_id = posts[0].id;
501 +
502 + // Mod removes the opening post via the HTTP endpoint.
503 + let mod_id = h.login_as("cascademod").await;
504 + h.add_membership(mod_id, comm_id, "moderator").await;
505 + let thread_url = format!("/p/test/general/{}", thread_id);
506 + h.client.get(&thread_url).await; // prime CSRF token
507 +
508 + let remove_url = format!("/p/test/general/{}/posts/{}/remove", thread_id, op_id);
509 + let resp = h.client.post_form(&remove_url, "").await;
510 + assert!(resp.status.is_redirection(), "Expected redirect, got {}", resp.status);
511 +
512 + // The whole thread is now soft-deleted.
513 + let deleted: (bool,) = sqlx::query_as("SELECT deleted_at IS NOT NULL FROM threads WHERE id = $1")
514 + .bind(thread_id)
515 + .fetch_one(&h.db)
516 + .await
517 + .unwrap();
518 + assert!(deleted.0, "Removing the OP should soft-delete the thread");
519 +
520 + // The thread page now 404s instead of leaving a headless thread.
521 + let resp = h.client.get(&thread_url).await;
522 + assert_eq!(resp.status.as_u16(), 404, "Cascaded thread should 404");
523 +
524 + // Replying to the cascaded thread is rejected (no live headless thread).
525 + let reply_resp = h
526 + .client
527 + .post_form(&format!("{}/reply", thread_url), "body=ghost+reply")
528 + .await;
529 + assert_eq!(
530 + reply_resp.status.as_u16(),
531 + 404,
532 + "Replying to a cascaded thread should 404"
533 + );
534 + }
535 +
536 + #[tokio::test]
537 + async fn removing_reply_does_not_cascade() {
538 + let mut h = TestHarness::new().await;
539 + let author_id = h.login_as("noncascadeauthor").await;
540 + let comm_id = h.create_community("Test", "test").await;
541 + let cat_id = h.create_category(comm_id, "General", "general").await;
542 + h.add_membership(author_id, comm_id, "member").await;
543 +
544 + let thread_id = h
545 + .create_thread_with_post(cat_id, author_id, "Survivor Thread", "Opening content")
546 + .await;
547 + mt_db::mutations::create_post(&h.db, thread_id, author_id, "A reply", "<p>A reply</p>")
548 + .await
549 + .unwrap();
550 +
551 + let posts = mt_db::queries::list_posts_in_thread(&h.db, thread_id)
552 + .await
553 + .unwrap();
554 + let reply_id = posts[1].id;
555 +
556 + // Mod removes the reply (not the OP).
557 + let mod_id = h.login_as("noncascademod").await;
558 + h.add_membership(mod_id, comm_id, "moderator").await;
559 + let thread_url = format!("/p/test/general/{}", thread_id);
560 + h.client.get(&thread_url).await;
561 +
562 + let remove_url = format!("/p/test/general/{}/posts/{}/remove", thread_id, reply_id);
563 + let resp = h.client.post_form(&remove_url, "").await;
564 + assert!(resp.status.is_redirection(), "Expected redirect, got {}", resp.status);
565 +
566 + // The thread survives: not deleted, still viewable.
567 + let deleted: (bool,) = sqlx::query_as("SELECT deleted_at IS NOT NULL FROM threads WHERE id = $1")
568 + .bind(thread_id)
569 + .fetch_one(&h.db)
570 + .await
571 + .unwrap();
572 + assert!(!deleted.0, "Removing a reply must not delete the thread");
573 +
574 + let resp = h.client.get(&thread_url).await;
575 + assert_eq!(resp.status.as_u16(), 200, "Thread should still be viewable");
576 + }
M multithreaded/tests/workflows/mutations.rs +62
@@ -739,3 +739,65 @@ async fn upsert_user_updates_on_conflict() {
739 739 assert_eq!(username, "newname");
740 740 assert_eq!(display_name.as_deref(), Some("New Display"));
741 741 }
742 +
743 + #[tokio::test]
744 + async fn s3_purge_sweep_targets_only_removed_unpurged_images() {
745 + let h = TestHarness::new().await;
746 + let comm_id = h.create_community("Test", "test").await;
747 +
748 + let uploader = Uuid::new_v4();
749 + let moderator = Uuid::new_v4();
750 + for (id, name) in [(uploader, "purgeuploader"), (moderator, "purgemod")] {
751 + sqlx::query("INSERT INTO users (mnw_account_id, username) VALUES ($1, $2)")
752 + .bind(id)
753 + .bind(name)
754 + .execute(&h.db)
755 + .await
756 + .unwrap();
757 + }
758 +
759 + // A live image and a removed image.
760 + let live = mt_db::mutations::insert_image(
761 + &h.db, uploader, comm_id, "s3/live.jpg", "live.jpg", "image/jpeg", 100,
762 + )
763 + .await
764 + .unwrap();
765 + let removed = mt_db::mutations::insert_image(
766 + &h.db, uploader, comm_id, "s3/removed.jpg", "removed.jpg", "image/jpeg", 100,
767 + )
768 + .await
769 + .unwrap();
770 + mt_db::mutations::remove_image(&h.db, removed, moderator)
771 + .await
772 + .unwrap();
773 +
774 + // Only the removed-and-unpurged image is pending purge.
775 + let pending = mt_db::queries::list_images_pending_s3_purge(&h.db, 100)
776 + .await
777 + .unwrap();
778 + let ids: Vec<Uuid> = pending.iter().map(|p| p.id).collect();
779 + assert!(ids.contains(&removed), "removed image must be pending purge");
780 + assert!(!ids.contains(&live), "live image must not be pending purge");
781 + assert_eq!(
782 + pending.iter().find(|p| p.id == removed).map(|p| p.s3_key.as_str()),
783 + Some("s3/removed.jpg"),
784 + "pending row must carry the S3 key for deletion"
785 + );
786 +
787 + // After marking purged, it drops out of the sweep — convergent.
788 + mt_db::mutations::mark_images_s3_purged(&h.db, &[removed])
789 + .await
790 + .unwrap();
791 + let pending = mt_db::queries::list_images_pending_s3_purge(&h.db, 100)
792 + .await
793 + .unwrap();
794 + assert!(
795 + !pending.iter().any(|p| p.id == removed),
796 + "purged image must not be revisited by the sweep"
797 + );
798 +
799 + // Marking an empty batch is a no-op (sweep's failed-batch path).
800 + mt_db::mutations::mark_images_s3_purged(&h.db, &[])
801 + .await
802 + .unwrap();
803 + }