Skip to main content

max / makenotwork

audit Run 16 Phase 3: Concurrency - begin_key_rotation: was a non-transactional check-then-insert; two devices could both observe "no active rotation" and both INSERT. Run it in a transaction and take SELECT ... FOR UPDATE on the sync_keys row so concurrent begins for the same (app, user) serialize (mirrors complete_key_rotation). - duplicate_item: replace the pre-check + counter loop (a slug TOCTOU that 500'd the loser on a raw 23505) with a savepoint-per-attempt INSERT that retries with a numeric suffix on a unique violation, so the DB's unique index arbitrates. Remove the now-dead item_slug_exists helper. - guest_checkout: the 23505 dedup branch is inert for guests (buyer_id is always NULL; NULLS DISTINCT). Guest flooding is already bounded by the endpoint's per-IP rate limit; document that accurately rather than implying the unique index dedups guests. True per-guest dedup needs a client idempotency key (left for a protocol decision). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-02 13:36 UTC
Commit: 557cb176c55ac59eccb995ca7b0f1f9668370652
Parent: 95a103e
4 files changed, +81 insertions, -62 deletions
M server/src/db/items/bulk.rs +54 -38
@@ -225,47 +225,63 @@ pub async fn duplicate_item(pool: &PgPool, source_id: ItemId, user_id: UserId) -
225 225 .await?;
226 226 let copy_title = format!("Copy of {}", &source.title);
227 227 let copy_title: String = copy_title.chars().take(200).collect();
228 - let mut slug = crate::helpers::slugify(&copy_title);
229 - if super::item_slug_exists(&mut *tx, source.project_id, &slug).await? {
230 - let base = slug.clone();
231 - let mut counter = 2u32;
232 - loop {
233 - if counter > 100 {
234 - return Err(crate::error::AppError::BadRequest(
235 - "Too many copies with similar names. Rename an existing copy first.".to_string(),
236 - ));
228 + let base_slug = crate::helpers::slugify(&copy_title).to_string();
229 +
230 + // Step 1: Clone item row with a collision-safe slug. A pre-check
231 + // (`item_slug_exists`) followed by an INSERT races two concurrent
232 + // duplicates onto the same slug — the loser hit a raw 23505 -> 500. Instead
233 + // attempt the INSERT under a savepoint and retry with a numeric suffix on a
234 + // unique violation, so the DB's unique index is the arbiter, not a
235 + // check-then-write window. Mirrors `helpers::insert_with_unique_slug`,
236 + // adapted to run inside this transaction: a bare 23505 aborts the tx, so
237 + // each attempt is wrapped in a savepoint and `ROLLBACK TO SAVEPOINT`
238 + // recovers the tx for the next try.
239 + let mut suffix = 1u32;
240 + let new_item = loop {
241 + let candidate = if suffix == 1 {
242 + base_slug.clone()
243 + } else {
244 + format!("{base_slug}-{suffix}")
245 + };
246 + sqlx::query("SAVEPOINT dup_item_slug").execute(&mut *tx).await?;
247 + let res = sqlx::query_as::<_, DbItem>(
248 + r#"
249 + INSERT INTO items (
250 + project_id, title, description, price_cents, item_type, thumbnail_url,
251 + sort_order, body, word_count, reading_time_minutes, duration_seconds,
252 + episode_number, enable_license_keys, default_max_activations,
253 + pwyw_enabled, pwyw_min_cents, is_public, slug
254 + )
255 + SELECT
256 + project_id, LEFT('Copy of ' || title, 200), description, price_cents,
257 + item_type, thumbnail_url, sort_order, body, word_count,
258 + reading_time_minutes, duration_seconds, episode_number,
259 + enable_license_keys, default_max_activations, pwyw_enabled,
260 + pwyw_min_cents, false, $2
261 + FROM items WHERE id = $1
262 + RETURNING *
263 + "#,
264 + )
265 + .bind(source_id)
266 + .bind(&candidate)
267 + .fetch_one(&mut *tx)
268 + .await;
269 + match res {
270 + Ok(item) => {
271 + sqlx::query("RELEASE SAVEPOINT dup_item_slug").execute(&mut *tx).await?;
272 + break item;
237 273 }
238 - slug = crate::db::validated_types::Slug::from_trusted(format!("{}-{}", base, counter));
239 - if !super::item_slug_exists(&mut *tx, source.project_id, &slug).await? {
240 - break;
274 + Err(e) => {
275 + sqlx::query("ROLLBACK TO SAVEPOINT dup_item_slug").execute(&mut *tx).await?;
276 + let app_err = crate::error::AppError::from(e);
277 + if crate::helpers::is_unique_violation(&app_err) && suffix < 100 {
278 + suffix += 1;
279 + continue;
280 + }
281 + return Err(app_err);
241 282 }
242 - counter += 1;
243 283 }
244 - }
245 -
246 - // Step 1: Clone item row
247 - let new_item = sqlx::query_as::<_, DbItem>(
248 - r#"
249 - INSERT INTO items (
250 - project_id, title, description, price_cents, item_type, thumbnail_url,
251 - sort_order, body, word_count, reading_time_minutes, duration_seconds,
252 - episode_number, enable_license_keys, default_max_activations,
253 - pwyw_enabled, pwyw_min_cents, is_public, slug
254 - )
255 - SELECT
256 - project_id, LEFT('Copy of ' || title, 200), description, price_cents,
257 - item_type, thumbnail_url, sort_order, body, word_count,
258 - reading_time_minutes, duration_seconds, episode_number,
259 - enable_license_keys, default_max_activations, pwyw_enabled,
260 - pwyw_min_cents, false, $2
261 - FROM items WHERE id = $1
262 - RETURNING *
263 - "#,
264 - )
265 - .bind(source_id)
266 - .bind(&slug)
267 - .fetch_one(&mut *tx)
268 - .await?;
284 + };
269 285
270 286 // Step 2: Copy tags
271 287 sqlx::query(
M server/src/db/items/mod.rs -17
@@ -102,23 +102,6 @@ pub async fn create_item(
102 102 .await
103 103 }
104 104
105 - /// Check whether a slug already exists for a given project.
106 - pub(super) async fn item_slug_exists<'e, E: sqlx::Executor<'e, Database = sqlx::Postgres>>(
107 - executor: E,
108 - project_id: ProjectId,
109 - slug: &super::validated_types::Slug,
110 - ) -> Result<bool> {
111 - let exists: bool = sqlx::query_scalar(
112 - "SELECT EXISTS(SELECT 1 FROM items WHERE project_id = $1 AND slug = $2)",
113 - )
114 - .bind(project_id)
115 - .bind(slug)
116 - .fetch_one(executor)
117 - .await?;
118 -
119 - Ok(exists)
120 - }
121 -
122 105 /// Fetch an item by primary key. Returns `None` if not found.
123 106 #[tracing::instrument(skip_all)]
124 107 pub async fn get_item_by_id(pool: &PgPool, id: ItemId) -> Result<Option<DbItem>> {
M server/src/db/synckit/rotation.rs +17 -7
@@ -20,13 +20,22 @@ pub async fn begin_key_rotation(
20 20 new_encrypted_key: &str,
21 21 expected_key_version: i32,
22 22 ) -> Result<std::result::Result<crate::db::models::DbSyncKeyRotation, &'static str>> {
23 - // Verify key_version matches
23 + // The whole begin is a check-then-insert: without serialization two devices
24 + // can both observe "no existing rotation" and both INSERT, racing on the
25 + // unique constraint (one 500s, or two rotations exist). Run it in a
26 + // transaction and take a row lock on the `sync_keys` row up front —
27 + // `FOR UPDATE` serializes every concurrent begin for the same (app, user),
28 + // so the existing-rotation check and the insert are atomic. (Compare
29 + // `complete_key_rotation`, which already locks.)
30 + let mut tx = pool.begin().await?;
31 +
32 + // Verify key_version matches, holding the key row for the rest of the tx.
24 33 let key_row: Option<(i32, i32)> = sqlx::query_as(
25 - "SELECT key_version, key_id FROM sync_keys WHERE app_id = $1 AND user_id = $2",
34 + "SELECT key_version, key_id FROM sync_keys WHERE app_id = $1 AND user_id = $2 FOR UPDATE",
26 35 )
27 36 .bind(app_id)
28 37 .bind(user_id)
29 - .fetch_optional(pool)
38 + .fetch_optional(&mut *tx)
30 39 .await?;
31 40
32 41 let Some((current_version, current_key_id)) = key_row else {
@@ -37,13 +46,13 @@ pub async fn begin_key_rotation(
37 46 return Ok(Err("key version mismatch"));
38 47 }
39 48
40 - // Check for existing rotation
49 + // Check for existing rotation (serialized by the lock above).
41 50 let existing = sqlx::query_as::<_, crate::db::models::DbSyncKeyRotation>(
42 51 "SELECT * FROM sync_key_rotations WHERE app_id = $1 AND user_id = $2",
43 52 )
44 53 .bind(app_id)
45 54 .bind(user_id)
46 - .fetch_optional(pool)
55 + .fetch_optional(&mut *tx)
47 56 .await?;
48 57
49 58 if let Some(rotation) = existing {
@@ -60,7 +69,7 @@ pub async fn begin_key_rotation(
60 69 )
61 70 .bind(app_id)
62 71 .bind(user_id)
63 - .fetch_one(pool)
72 + .fetch_one(&mut *tx)
64 73 .await?;
65 74
66 75 let new_key_id = current_key_id + 1;
@@ -79,9 +88,10 @@ pub async fn begin_key_rotation(
79 88 .bind(current_version)
80 89 .bind(new_key_id)
81 90 .bind(target_seq)
82 - .fetch_one(pool)
91 + .fetch_one(&mut *tx)
83 92 .await?;
84 93
94 + tx.commit().await?;
85 95 Ok(Ok(rotation))
86 96 }
87 97
M server/src/routes/api/guest_checkout.rs +10
@@ -213,6 +213,16 @@ pub(super) async fn create_guest_checkout(
213 213 },
214 214 ).await {
215 215 Ok(_) => {}
216 + // 23505 backstop. Note: the pending-checkout partial unique index is
217 + // `(buyer_id, item_id) WHERE status = 'pending'`, and a guest row always
218 + // has `buyer_id = NULL` (NULLS DISTINCT), so this branch does NOT dedup
219 + // rapid guest re-submits the way it does for the authenticated path —
220 + // guests have no stable identity at creation time (guest_email arrives
221 + // later, from the Stripe webhook). Guest checkout flooding is instead
222 + // bounded by the endpoint's per-IP rate limit (GUEST_CHECKOUT_RATE_LIMIT_*
223 + // in api/mod.rs). This branch therefore only fires on the effectively
224 + // impossible stripe_checkout_session_id collision; true per-guest dedup
225 + // would need a client-supplied idempotency key (a protocol change).
216 226 Err(AppError::Database(sqlx::Error::Database(ref db_err)))
217 227 if db_err.code().as_deref() == Some("23505") =>
218 228 {