Skip to main content

max / makenotwork

mt: make username reconcile and post creation atomic (ultra-fuzz S2 + M-St1) S2: a login that renames into a username a stale mirror row still holds tripped the users_username_key unique index and locked the user out. Both the OAuth login upsert and internal upsert_user now run a two-step transaction — vacate_stale_username frees the name from any other row (renaming it to a collision-proof mnw_<account-id> placeholder), then upserts. Two statements, not one CTE: a non-deferrable unique index is checked per-row, so the freed name wouldn't be visible to a same-statement insert. M-St1: thread/reply creation, insert_mentions, and set_thread_tags committed in separate statements, so a mid-sequence failure dropped mentions (no notification) and tags. Add *_tx variants taking a caller-owned connection (pool wrappers retained for tests) and fold all of them into one handler transaction. MN3 verified already mitigated: the upload route carries DefaultBodyLimit, so multipart field reads are bounded. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 06:17 UTC
Commit: ae9d34aacf1a17af579263a8a9bf79b83c45a607
Parent: aa8b724
5 files changed, +241 insertions, -66 deletions
A multithreaded/.sqlx/query-fede228e8a0209fde5f50a26baf05d01fc5a1319e1cc61386428f294979d8cf2.json +15
@@ -0,0 +1,15 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "UPDATE users\n SET username = 'mnw_' || mnw_account_id::text, updated_at = now()\n WHERE username = $1 AND mnw_account_id <> $2",
4 + "describe": {
5 + "columns": [],
6 + "parameters": {
7 + "Left": [
8 + "Text",
9 + "Uuid"
10 + ]
11 + },
12 + "nullable": []
13 + },
14 + "hash": "fede228e8a0209fde5f50a26baf05d01fc5a1319e1cc61386428f294979d8cf2"
15 + }
M multithreaded/crates/mt-db/src/mutations.rs +103 -13
@@ -34,6 +34,8 @@ pub async fn upsert_user(
34 34 username: &str,
35 35 display_name: Option<&str>,
36 36 ) -> Result<(), sqlx::Error> {
37 + let mut tx = pool.begin().await?;
38 + vacate_stale_username(&mut tx, mnw_account_id, username).await?;
37 39 sqlx::query!(
38 40 "INSERT INTO users (mnw_account_id, username, display_name)
39 41 VALUES ($1, $2, $3)
@@ -43,11 +45,49 @@ pub async fn upsert_user(
43 45 username,
44 46 display_name,
45 47 )
46 - .execute(pool)
48 + .execute(&mut *tx)
49 + .await?;
50 + tx.commit().await?;
51 + Ok(())
52 + }
53 +
54 + /// Free `username` from any *other* user row that still holds it, so the caller's
55 + /// upsert can claim it without tripping the `users_username_key` unique index.
56 + ///
57 + /// MNW is the source of truth for usernames and they are reusable there; a local
58 + /// mirror row holding a name its MNW account no longer owns is stale and must
59 + /// yield. Run this on the same transaction as the upsert (a single statement
60 + /// can't, because a non-deferrable unique index is checked per-row and the
61 + /// rename wouldn't yet be visible to the insert). The vacated row keeps its data
62 + /// under a collision-proof `mnw_<account-id>` placeholder until that user's own
63 + /// next login resets it. Closes the username-collision login lockout (S2).
64 + async fn vacate_stale_username(
65 + conn: &mut sqlx::PgConnection,
66 + mnw_account_id: Uuid,
67 + username: &str,
68 + ) -> Result<(), sqlx::Error> {
69 + sqlx::query!(
70 + "UPDATE users
71 + SET username = 'mnw_' || mnw_account_id::text, updated_at = now()
72 + WHERE username = $1 AND mnw_account_id <> $2",
73 + username,
74 + mnw_account_id,
75 + )
76 + .execute(&mut *conn)
47 77 .await?;
48 78 Ok(())
49 79 }
50 80
81 + /// Public wrapper over [`vacate_stale_username`] for the OAuth login upsert in the
82 + /// web crate, which builds its INSERT with a runtime query.
83 + pub async fn vacate_username_for_login(
84 + conn: &mut sqlx::PgConnection,
85 + mnw_account_id: Uuid,
86 + username: &str,
87 + ) -> Result<(), sqlx::Error> {
88 + vacate_stale_username(conn, mnw_account_id, username).await
89 + }
90 +
51 91 /// Ensure a user has a membership in a community with the given role.
52 92 /// Creates membership if none exists, does nothing if already a member.
53 93 #[tracing::instrument(skip_all)]
@@ -175,7 +215,28 @@ pub async fn create_thread_with_op(
175 215 body_html: &str,
176 216 ) -> Result<(Uuid, Uuid), sqlx::Error> {
177 217 let mut tx = pool.begin().await?;
218 + let ids =
219 + create_thread_with_op_tx(&mut tx, category_id, author_id, title, body_markdown, body_html)
220 + .await?;
221 + tx.commit().await?;
222 + Ok(ids)
223 + }
178 224
225 + /// Create a thread and its opening post on the caller's transaction.
226 + ///
227 + /// Lets a handler fold the thread, its OP, its mentions, and its tags into one
228 + /// atomic unit so a mid-sequence failure can't leave a thread whose @mentions or
229 + /// tags silently vanished (ultra-fuzz M-St1). The pool-taking
230 + /// [`create_thread_with_op`] wraps this in its own transaction.
231 + #[tracing::instrument(skip_all)]
232 + pub async fn create_thread_with_op_tx(
233 + conn: &mut sqlx::PgConnection,
234 + category_id: Uuid,
235 + author_id: Uuid,
236 + title: &str,
237 + body_markdown: &str,
238 + body_html: &str,
239 + ) -> Result<(Uuid, Uuid), sqlx::Error> {
179 240 let thread_id = sqlx::query_scalar!(
180 241 "INSERT INTO threads (category_id, author_id, title)
181 242 VALUES ($1, $2, $3)
@@ -184,7 +245,7 @@ pub async fn create_thread_with_op(
184 245 author_id,
185 246 title,
186 247 )
187 - .fetch_one(&mut *tx)
248 + .fetch_one(&mut *conn)
188 249 .await?;
189 250
190 251 let post_id = sqlx::query_scalar!(
@@ -196,10 +257,9 @@ pub async fn create_thread_with_op(
196 257 body_markdown,
197 258 body_html,
198 259 )
199 - .fetch_one(&mut *tx)
260 + .fetch_one(&mut *conn)
200 261 .await?;
201 262
202 - tx.commit().await?;
203 263 Ok((thread_id, post_id))
204 264 }
205 265
@@ -218,7 +278,25 @@ pub async fn create_post(
218 278 body_html: &str,
219 279 ) -> Result<Uuid, sqlx::Error> {
220 280 let mut tx = pool.begin().await?;
281 + let post_id = create_post_tx(&mut tx, thread_id, author_id, body_markdown, body_html).await?;
282 + tx.commit().await?;
283 + Ok(post_id)
284 + }
221 285
286 + /// Insert a reply and bump `last_activity_at` on the caller's transaction.
287 + ///
288 + /// Lets a handler fold the reply *and its mentions* into one atomic unit so a
289 + /// mid-sequence failure can't leave a committed reply whose @mentions silently
290 + /// vanished (ultra-fuzz M-St1). The pool-taking [`create_post`] wraps this in
291 + /// its own transaction for the simple, mention-free callers (and tests).
292 + #[tracing::instrument(skip_all)]
293 + pub async fn create_post_tx(
294 + conn: &mut sqlx::PgConnection,
295 + thread_id: Uuid,
296 + author_id: Uuid,
297 + body_markdown: &str,
298 + body_html: &str,
299 + ) -> Result<Uuid, sqlx::Error> {
222 300 let post_id = sqlx::query_scalar!(
223 301 "INSERT INTO posts (thread_id, author_id, body_markdown, body_html)
224 302 VALUES ($1, $2, $3, $4)
@@ -228,14 +306,13 @@ pub async fn create_post(
228 306 body_markdown,
229 307 body_html,
230 308 )
231 - .fetch_one(&mut *tx)
309 + .fetch_one(&mut *conn)
232 310 .await?;
233 311
234 312 sqlx::query!("UPDATE threads SET last_activity_at = now() WHERE id = $1", thread_id)
235 - .execute(&mut *tx)
313 + .execute(&mut *conn)
236 314 .await?;
237 315
238 - tx.commit().await?;
239 316 Ok(post_id)
240 317 }
241 318
@@ -989,8 +1066,22 @@ pub async fn set_thread_tags(
989 1066 tag_ids: &[Uuid],
990 1067 ) -> Result<(), sqlx::Error> {
991 1068 let mut tx = pool.begin().await?;
1069 + set_thread_tags_tx(&mut tx, thread_id, tag_ids).await?;
1070 + tx.commit().await?;
1071 + Ok(())
1072 + }
1073 +
1074 + /// Set a thread's tags on the caller's transaction, so a handler can fold tag
1075 + /// assignment into the same atomic unit as the thread/OP creation (M-St1). The
1076 + /// pool-taking [`set_thread_tags`] wraps this in its own transaction.
1077 + #[tracing::instrument(skip_all)]
1078 + pub async fn set_thread_tags_tx(
1079 + conn: &mut sqlx::PgConnection,
1080 + thread_id: Uuid,
1081 + tag_ids: &[Uuid],
1082 + ) -> Result<(), sqlx::Error> {
992 1083 sqlx::query!("DELETE FROM thread_tags WHERE thread_id = $1", thread_id)
993 - .execute(&mut *tx)
1084 + .execute(&mut *conn)
994 1085 .await?;
995 1086 if !tag_ids.is_empty() {
996 1087 // runtime-checked: dynamic SQL (cannot use compile-time macro)
@@ -1007,9 +1098,8 @@ pub async fn set_thread_tags(
1007 1098 for tag_id in tag_ids {
1008 1099 query = query.bind(tag_id);
1009 1100 }
1010 - query.execute(&mut *tx).await?;
1101 + query.execute(&mut *conn).await?;
1011 1102 }
1012 - tx.commit().await?;
1013 1103 Ok(())
1014 1104 }
1015 1105
@@ -1113,8 +1203,8 @@ pub async fn insert_link_preview(
1113 1203
1114 1204 /// Insert mention rows for a post (batch insert). Ignores duplicates.
1115 1205 #[tracing::instrument(skip_all)]
1116 - pub async fn insert_mentions(
1117 - pool: &PgPool,
1206 + pub async fn insert_mentions<'e, E: sqlx::PgExecutor<'e>>(
1207 + executor: E,
1118 1208 post_id: Uuid,
1119 1209 user_ids: &[Uuid],
1120 1210 ) -> Result<(), sqlx::Error> {
@@ -1135,7 +1225,7 @@ pub async fn insert_mentions(
1135 1225 for user_id in user_ids {
1136 1226 query = query.bind(user_id);
1137 1227 }
1138 - query.execute(pool).await?;
1228 + query.execute(executor).await?;
1139 1229 Ok(())
1140 1230 }
1141 1231
M multithreaded/src/auth.rs +29 -16
@@ -624,22 +624,35 @@ pub async fn callback(
624 624 // Upsert local user. `is_fan_plus`/`is_creator` are denormalised here so
625 625 // post rendering can look up the post author's perks via JOIN — see
626 626 // migration 026.
627 - let upsert_result = sqlx::query(
628 - r#"
629 - INSERT INTO users (mnw_account_id, username, display_name, avatar_url, is_fan_plus, is_creator)
630 - VALUES ($1, $2, $3, $4, $5, $6)
631 - ON CONFLICT (mnw_account_id) DO UPDATE
632 - SET username = $2, display_name = $3, avatar_url = $4,
633 - is_fan_plus = $5, is_creator = $6, updated_at = now()
634 - "#,
635 - )
636 - .bind(info.user_id)
637 - .bind(&info.username)
638 - .bind(&info.display_name)
639 - .bind(&info.avatar_url)
640 - .bind(info.perks.fan_plus)
641 - .bind(info.perks.is_creator)
642 - .execute(&state.db)
627 + //
628 + // First vacate the username from any stale mirror row that still holds it:
629 + // MNW usernames are mutable/reusable, so a login that renames into a name an
630 + // old row carries would otherwise trip the `users_username_key` unique index
631 + // and lock the user out (S2). The vacate and the upsert run on one
632 + // transaction so the freed name is visible to the insert.
633 + let upsert_result = async {
634 + let mut tx = state.db.begin().await?;
635 + mt_db::mutations::vacate_username_for_login(&mut tx, info.user_id, &info.username).await?;
636 + sqlx::query(
637 + r#"
638 + INSERT INTO users (mnw_account_id, username, display_name, avatar_url, is_fan_plus, is_creator)
639 + VALUES ($1, $2, $3, $4, $5, $6)
640 + ON CONFLICT (mnw_account_id) DO UPDATE
641 + SET username = $2, display_name = $3, avatar_url = $4,
642 + is_fan_plus = $5, is_creator = $6, updated_at = now()
643 + "#,
644 + )
645 + .bind(info.user_id)
646 + .bind(&info.username)
647 + .bind(&info.display_name)
648 + .bind(&info.avatar_url)
649 + .bind(info.perks.fan_plus)
650 + .bind(info.perks.is_creator)
651 + .execute(&mut *tx)
652 + .await?;
653 + tx.commit().await?;
654 + Ok::<(), sqlx::Error>(())
655 + }
643 656 .await;
644 657
645 658 if let Err(e) = upsert_result {
M multithreaded/src/routes/forum/posts.rs +40 -37
@@ -282,43 +282,44 @@ pub(in crate::routes) async fn create_thread_handler(
282 282 &state.db, body, community.id, &slug, user.user_id, author_plus,
283 283 ).await?;
284 284
285 - let (thread_id, post_id) =
286 - mt_db::mutations::create_thread_with_op(&state.db, category_id, user.user_id, title, body, &body_html)
287 - .await
288 - .map_err(|e| {
289 - tracing::error!(error = ?e, "db error creating thread");
290 - StatusCode::INTERNAL_SERVER_ERROR.into_response()
291 - })?;
285 + let tag_ids: Vec<uuid::Uuid> = form
286 + .tags
287 + .iter()
288 + .filter_map(|t| uuid::Uuid::parse_str(t).ok())
289 + .collect();
292 290
293 - if !mention_ids.is_empty() {
294 - mt_db::mutations::insert_mentions(&state.db, post_id, &mention_ids)
291 + // Create the thread, its opening post, its @mentions, and its tags as one
292 + // atomic unit. Previously these committed in separate statements, so a
293 + // failure after the thread insert silently dropped the mentions (no
294 + // notification) and tags (M-St1).
295 + let mut tx = begin_tx(&state.db).await?;
296 + let (thread_id, post_id) = mt_db::mutations::create_thread_with_op_tx(
297 + &mut tx, category_id, user.user_id, title, body, &body_html,
298 + )
299 + .await
300 + .map_err(|e| {
301 + tracing::error!(error = ?e, "db error creating thread");
302 + StatusCode::INTERNAL_SERVER_ERROR.into_response()
303 + })?;
304 + mt_db::mutations::insert_mentions(&mut *tx, post_id, &mention_ids)
305 + .await
306 + .map_err(|e| {
307 + tracing::error!(error = ?e, "db error inserting mentions");
308 + StatusCode::INTERNAL_SERVER_ERROR.into_response()
309 + })?;
310 + if !tag_ids.is_empty() {
311 + mt_db::mutations::set_thread_tags_tx(&mut tx, thread_id, &tag_ids)
295 312 .await
296 313 .map_err(|e| {
297 - tracing::error!(error = ?e, "db error inserting mentions");
314 + tracing::error!(error = ?e, "db error setting thread tags");
298 315 StatusCode::INTERNAL_SERVER_ERROR.into_response()
299 316 })?;
300 317 }
318 + commit_tx(tx).await?;
301 319
302 320 // Fetch link previews in background (best-effort, never blocks response)
303 321 spawn_link_preview_fetch(state.clone(), body.to_string(), post_id);
304 322
305 - // Save tags if any were selected
306 - if !form.tags.is_empty() {
307 - let tag_ids: Vec<uuid::Uuid> = form
308 - .tags
309 - .iter()
310 - .filter_map(|t| uuid::Uuid::parse_str(t).ok())
311 - .collect();
312 - if !tag_ids.is_empty() {
313 - mt_db::mutations::set_thread_tags(&state.db, thread_id, &tag_ids)
314 - .await
315 - .map_err(|e| {
316 - tracing::error!(error = ?e, "db error setting thread tags");
317 - StatusCode::INTERNAL_SERVER_ERROR.into_response()
318 - })?;
319 - }
320 - }
321 -
322 323 Ok(Redirect::to(&format!(
323 324 "/p/{slug}/{category_slug}/{thread_id}?toast=Thread+created"
324 325 )))
@@ -365,21 +366,23 @@ pub(in crate::routes) async fn create_reply_handler(
365 366 ).await?;
366 367
367 368 let thread_id = parse_uuid(&thread_id_str)?;
368 - let post_id = mt_db::mutations::create_post(&state.db, thread_id, user.user_id, body, &body_html)
369 +
370 + // Create the reply and its @mentions as one atomic unit so a failure between
371 + // them can't leave a committed reply whose mentions silently vanished (M-St1).
372 + let mut tx = begin_tx(&state.db).await?;
373 + let post_id = mt_db::mutations::create_post_tx(&mut tx, thread_id, user.user_id, body, &body_html)
369 374 .await
370 375 .map_err(|e| {
371 376 tracing::error!(error = ?e, "db error creating reply");
372 377 StatusCode::INTERNAL_SERVER_ERROR.into_response()
373 378 })?;
374 -
375 - if !mention_ids.is_empty() {
376 - mt_db::mutations::insert_mentions(&state.db, post_id, &mention_ids)
377 - .await
378 - .map_err(|e| {
379 - tracing::error!(error = ?e, "db error inserting mentions");
380 - StatusCode::INTERNAL_SERVER_ERROR.into_response()
381 - })?;
382 - }
379 + mt_db::mutations::insert_mentions(&mut *tx, post_id, &mention_ids)
380 + .await
381 + .map_err(|e| {
382 + tracing::error!(error = ?e, "db error inserting mentions");
383 + StatusCode::INTERNAL_SERVER_ERROR.into_response()
384 + })?;
385 + commit_tx(tx).await?;
383 386
384 387 // Fetch link previews in background (best-effort, never blocks response)
385 388 spawn_link_preview_fetch(state.clone(), body.to_string(), post_id);
M multithreaded/tests/workflows/mutations.rs +54
@@ -880,3 +880,57 @@ async fn s3_purge_sweep_targets_only_removed_unpurged_images() {
880 880 .await
881 881 .unwrap();
882 882 }
883 +
884 + // ============================================================================
885 + // Username-collision reconciliation (ultra-fuzz S2)
886 + // ============================================================================
887 +
888 + /// A login that renames into a username a *stale* mirror row still holds must
889 + /// succeed, not 23505-lockout. The stale row yields the name (MNW is the source
890 + /// of truth; usernames are reusable there) and keeps its data under a
891 + /// collision-proof `mnw_<account-id>` placeholder.
892 + #[tokio::test]
893 + async fn upsert_user_reclaims_username_from_stale_row() {
894 + let h = TestHarness::new().await;
895 + let old = Uuid::new_v4();
896 + let new = Uuid::new_v4();
897 +
898 + // `old` currently owns "shared".
899 + mt_db::mutations::upsert_user(&h.db, old, "shared", None).await.unwrap();
900 +
901 + // `new` logs in having taken the name "shared" upstream. This must not error.
902 + mt_db::mutations::upsert_user(&h.db, new, "shared", None)
903 + .await
904 + .expect("rename into a name a stale row holds must not lock the user out");
905 +
906 + // `new` now owns "shared"; `old` was vacated to its placeholder.
907 + let owner: Uuid = sqlx::query_scalar("SELECT mnw_account_id FROM users WHERE username = 'shared'")
908 + .fetch_one(&h.db)
909 + .await
910 + .unwrap();
911 + assert_eq!(owner, new, "the current login must own the reused username");
912 +
913 + let old_name: String = sqlx::query_scalar("SELECT username FROM users WHERE mnw_account_id = $1")
914 + .bind(old)
915 + .fetch_one(&h.db)
916 + .await
917 + .unwrap();
918 + assert_eq!(old_name, format!("mnw_{old}"), "stale row keeps its data under a placeholder name");
919 + }
920 +
921 + /// Re-upserting the same account with its own unchanged username is a no-op rename
922 + /// (the `mnw_account_id <> $2` guard skips self), not a self-collision.
923 + #[tokio::test]
924 + async fn upsert_user_same_account_same_name_is_stable() {
925 + let h = TestHarness::new().await;
926 + let acct = Uuid::new_v4();
927 + mt_db::mutations::upsert_user(&h.db, acct, "stable", Some("Display")).await.unwrap();
928 + mt_db::mutations::upsert_user(&h.db, acct, "stable", Some("Display Two")).await.unwrap();
929 +
930 + let name: String = sqlx::query_scalar("SELECT username FROM users WHERE mnw_account_id = $1")
931 + .bind(acct)
932 + .fetch_one(&h.db)
933 + .await
934 + .unwrap();
935 + assert_eq!(name, "stable");
936 + }