diff --git a/backend/migrations/022_client_upload_idempotency.down.sql b/backend/migrations/022_client_upload_idempotency.down.sql new file mode 100644 index 0000000..e803fb9 --- /dev/null +++ b/backend/migrations/022_client_upload_idempotency.down.sql @@ -0,0 +1,2 @@ +DROP INDEX IF EXISTS upload_client_upload_id_key; +ALTER TABLE upload DROP COLUMN IF EXISTS client_upload_id; diff --git a/backend/migrations/022_client_upload_idempotency.up.sql b/backend/migrations/022_client_upload_idempotency.up.sql new file mode 100644 index 0000000..e3c71bf --- /dev/null +++ b/backend/migrations/022_client_upload_idempotency.up.sql @@ -0,0 +1,25 @@ +-- Idempotency key for uploads, supplied by the client. +-- +-- The failure this closes is the ordinary one on a phone, not an exotic race: the server +-- receives the body, validates it, commits the row, and the response is lost on the way back +-- because the guest walked out of range or the AP dropped the connection. The client sees a +-- network error with the blob still in hand, marks the item retryable, and re-sends it — both +-- when the guest taps "Erneut" and automatically when the queue requeues on reconnect. Every +-- attempt minted a fresh `Uuid::new_v4()` server-side, so the same photo landed in the gallery +-- two or three times and was charged against the guest's storage quota each time. +-- +-- The client already has a stable per-queue-item UUID, so it costs nothing to send. NULL is +-- allowed and unconstrained: uploads that predate this column, and any client that doesn't send +-- one, keep working exactly as before. +ALTER TABLE upload ADD COLUMN client_upload_id UUID; + +-- Partial rather than a plain UNIQUE. Postgres would tolerate the NULLs either way, but indexing +-- only the rows that carry a key keeps it small and states the rule exactly: uniqueness applies +-- where a key exists, and nowhere else. +-- +-- Scoped globally rather than per user or per event. The key is a client-generated v4 UUID, so a +-- collision between two different photos is not a real possibility, and a single-column index +-- means the uniqueness check cannot be wrong about which event or user a retry belongs to. +CREATE UNIQUE INDEX upload_client_upload_id_key + ON upload (client_upload_id) + WHERE client_upload_id IS NOT NULL; diff --git a/backend/src/handlers/upload.rs b/backend/src/handlers/upload.rs index 0422e12..06a820a 100644 --- a/backend/src/handlers/upload.rs +++ b/backend/src/handlers/upload.rs @@ -111,6 +111,9 @@ pub async fn upload( let mut streamed: Option<(i64, Vec)> = None; // (size, head bytes for sniffing) let mut caption: Option = None; let mut hashtags_csv: Option = None; + // The client's idempotency key. Optional: an older client, or any other caller, simply + // doesn't send one and gets the previous behaviour. + let mut client_upload_id: Option = None; // Wrap the multipart read so any error after the temp file is created still cleans // it up (a mid-stream parse failure must not leave a stray `.tmp` on disk). @@ -158,6 +161,16 @@ pub async fn upload( .map_err(|e| AppError::BadRequest(e.to_string()))?, ); } + "client_upload_id" => { + let raw = field + .text() + .await + .map_err(|e| AppError::BadRequest(e.to_string()))?; + // A malformed key is not worth rejecting an upload over — the photo is the + // thing the guest cares about. Drop the key and lose only the retry + // protection, which is exactly where we were before it existed. + client_upload_id = Uuid::parse_str(raw.trim()).ok(); + } _ => {} } } @@ -170,6 +183,29 @@ pub async fn upload( return Err(e); } + // Idempotency, fast path: this key already has a live upload, so the previous attempt DID + // succeed and only its response was lost. Replay that response instead of storing the photo + // a second time and charging the guest's quota twice. + // + // The body has necessarily already been streamed to disk — the key arrives as a multipart + // field, so it cannot be known before the body is read. Re-sending the bytes is the client's + // cost and it has already been paid by the time we get here; what has to be prevented is a + // second ROW, and that is what this does. The concurrent case (two retries in flight at once) + // is caught by the unique index inside the transaction below. + if let Some(cid) = client_upload_id + && let Some(existing) = Upload::find_by_client_upload_id(&state.pool, auth.user_id, cid) + .await + .map_err(AppError::from)? + { + let _ = tokio::fs::remove_file(&temp_abs).await; + tracing::info!( + client_upload_id = %cid, upload_id = %existing.id, + "duplicate upload suppressed; replaying the original response" + ); + let dto = replay_upload_dto(&state, &existing, &user.display_name).await; + return Ok((StatusCode::OK, Json(dto))); + } + // From here on the temp file may exist; every validation failure removes it before // returning so a rejected upload never leaves bytes behind. let (size, head) = match streamed { @@ -239,18 +275,42 @@ pub async fn upload( // vanish with, at best, a vague "could not be processed". Rejecting here gives them a // reason at the door that they can act on, and it uses the SAME budget the worker // enforces, so admission and processing cannot disagree. - if mime.starts_with("image/") && crate::services::imaging::exceeds_decode_budget(&temp_abs) { - let mp = crate::services::imaging::megapixels(&temp_abs); - tracing::info!( - %mime, megapixels = ?mp, - "rejecting an image that exceeds the decode budget at admission" - ); - let _ = tokio::fs::remove_file(&temp_abs).await; - let detail = mp.map_or(String::new(), |mp| format!(" (ca. {mp:.0} Megapixel)")); - return Err(AppError::BadRequest(format!( - "Bild hat zu viele Bildpunkte{detail} und kann nicht verarbeitet werden. \ - Bitte verkleinere es und lade es erneut hoch." - ))); + // + // Both probes open the file and run the codec's header parse — synchronous filesystem and + // CPU work. They ran inline on the async task, which on this 2-vCPU box means tokio has + // exactly two worker threads and every upload stalled half the runtime's request-serving + // capacity. Everything else in the app that blocks (image encode, bcrypt) is already on the + // blocking pool; this was the one that wasn't. + if mime.starts_with("image/") { + let probe_path = temp_abs.clone(); + let probe = tokio::task::spawn_blocking(move || { + let over = crate::services::imaging::exceeds_decode_budget(&probe_path); + // Only pay for the second header read when it will actually be shown to the guest. + let mp = over + .then(|| crate::services::imaging::megapixels(&probe_path)) + .flatten(); + (over, mp) + }) + .await; + // A join error is the blocking pool panicking or shutting down. That says nothing about + // the image, so admit it and let the compression worker be the judge rather than + // rejecting a photo for an infrastructure reason. + let (over_budget, mp) = probe.unwrap_or_else(|e| { + tracing::warn!(error = ?e, "decode-budget probe failed to run; admitting the upload"); + (false, None) + }); + if over_budget { + tracing::info!( + %mime, megapixels = ?mp, + "rejecting an image that exceeds the decode budget at admission" + ); + let _ = tokio::fs::remove_file(&temp_abs).await; + let detail = mp.map_or(String::new(), |mp| format!(" (ca. {mp:.0} Megapixel)")); + return Err(AppError::BadRequest(format!( + "Bild hat zu viele Bildpunkte{detail} und kann nicht verarbeitet werden. \ + Bitte verkleinere es und lade es erneut hoch." + ))); + } } // Per-user storage quota — dynamic formula based on available disk space and the @@ -273,7 +333,13 @@ pub async fn upload( if prospective_total > limit { let _ = tokio::fs::remove_file(&temp_abs).await; return Err(AppError::QuotaExceeded( - "Du hast dein Upload-Limit für dieses Event erreicht.".into(), + // Name the remedy, because the guest cannot see the number. Every quota + // display is staff-gated by design, so a guest hitting this had no idea what + // the limit was, how close they were, or what to do — and the one sentence + // that tells them ("delete older posts") lived inside the staff-only block. + "Du hast dein Upload-Limit für dieses Event erreicht. Lösche ältere eigene \ + Beiträge, um wieder Platz zu schaffen." + .into(), )); } } @@ -364,10 +430,20 @@ pub async fn upload( }; if inc.rows_affected() == 0 { return Err(AppError::QuotaExceeded( - "Du hast dein Upload-Limit für dieses Event erreicht.".into(), + // Name the remedy, because the guest cannot see the number. Every quota + // display is staff-gated by design, so a guest hitting this had no idea what + // the limit was, how close they were, or what to do — and the one sentence + // that tells them ("delete older posts") lived inside the staff-only block. + "Du hast dein Upload-Limit für dieses Event erreicht. Lösche ältere eigene \ + Beiträge, um wieder Platz zu schaffen." + .into(), )); } - let upload = Upload::create( + // `None` means a concurrent request already stored this key. The transaction — quota + // increment included — is abandoned by returning here, and the caller replays the winning + // row. This is the narrow race the fast path above cannot see: two retries of the same + // photo in flight at the same moment. + let Some(upload) = Upload::create( &mut *tx, auth.event_id, auth.user_id, @@ -375,8 +451,12 @@ pub async fn upload( &mime, size, caption.as_deref(), + client_upload_id, ) - .await?; + .await? + else { + return Err(AppError::Conflict(DUPLICATE_UPLOAD_MARKER.into())); + }; for tag in &tags { let h = Hashtag::upsert(&mut *tx, auth.event_id, tag).await?; Hashtag::link_to_upload(&mut *tx, upload.id, h.id).await?; @@ -390,6 +470,28 @@ pub async fn upload( // row will ever reference it, so remove it now rather than orphan bytes on disk. let upload = match tx_result { Ok(u) => u, + // The concurrent duplicate resolved inside the transaction. The winner's row is committed; + // answer with it so both retries of the same photo get the same successful reply. + Err(AppError::Conflict(ref marker)) if marker == DUPLICATE_UPLOAD_MARKER => { + let _ = tokio::fs::remove_file(&absolute_path).await; + let existing = match client_upload_id { + Some(cid) => Upload::find_by_client_upload_id(&state.pool, auth.user_id, cid) + .await + .map_err(AppError::from)?, + None => None, + }; + // If the winning row has vanished between the conflict and this lookup (deleted in + // the intervening milliseconds), there is nothing to replay — report the conflict. + let existing = existing.ok_or_else(|| { + AppError::Conflict("Dieser Upload wurde bereits verarbeitet.".into()) + })?; + tracing::info!( + upload_id = %existing.id, + "concurrent duplicate upload resolved; replaying the stored row" + ); + let dto = replay_upload_dto(&state, &existing, &user.display_name).await; + return Ok((StatusCode::OK, Json(dto))); + } Err(e) => { let _ = tokio::fs::remove_file(&absolute_path).await; return Err(e); @@ -466,7 +568,15 @@ pub async fn edit_upload( } if let Some(ref hashtags) = body.hashtags { Hashtag::unlink_all_from_upload(&mut *tx, upload_id).await?; - for tag in hashtags { + // Sort + dedup before upserting, exactly as the upload path does. `Hashtag::upsert` + // takes row locks, so two transactions touching the same two tags in OPPOSITE order + // deadlock; Postgres aborts one after ~1s and the guest gets a 500. Here the order is + // whatever the client sent, so it is genuinely attacker-free but genuinely unordered. + // Sort on the NORMALISED form — that is the key `upsert` actually locks on. + let mut tags: Vec<&String> = hashtags.iter().collect(); + tags.sort_by_key(|t| t.trim().trim_start_matches('#').to_lowercase()); + tags.dedup_by_key(|t| t.trim().trim_start_matches('#').to_lowercase()); + for tag in tags { let h = Hashtag::upsert(&mut *tx, auth.event_id, tag).await?; Hashtag::link_to_upload(&mut *tx, upload_id, h.id).await?; } @@ -592,6 +702,69 @@ async fn stream_field_to_file( Ok((total as i64, head)) } +/// Sentinel for the duplicate detected INSIDE the commit transaction. It never reaches a client: +/// the caller intercepts this exact `Conflict` and answers with the stored row. A marker rather +/// than a new `AppError` variant because the condition is local to this one handler and returning +/// early is the only way to abandon the transaction from inside the async block. +const DUPLICATE_UPLOAD_MARKER: &str = "__duplicate_client_upload_id__"; + +/// Rebuild the response for an upload that already exists, so a retry is answered exactly as the +/// original was. +/// +/// Reads the live state rather than assuming a fresh row: by the time a retry arrives — a +/// reconnect can be minutes later — the derivatives may have been generated and the photo may +/// already have been liked, and a response claiming otherwise would be wrong in a way the client +/// has no way to detect. +/// +/// Every read here fails soft. This is the success path of an upload that is already safely +/// stored; degrading to a sparser response is fine, failing the request is not. +async fn replay_upload_dto(state: &AppState, upload: &Upload, uploader_name: &str) -> UploadDto { + let hashtags: Vec = sqlx::query_scalar( + "SELECT h.tag FROM upload_hashtag uh + JOIN hashtag h ON h.id = uh.hashtag_id + WHERE uh.upload_id = $1 + ORDER BY h.tag", + ) + .bind(upload.id) + .fetch_all(&state.pool) + .await + .unwrap_or_default(); + + let counts: Option<(i64, i64, bool)> = sqlx::query_as( + "SELECT v.like_count, v.comment_count, + EXISTS (SELECT 1 FROM \"like\" l WHERE l.upload_id = v.id AND l.user_id = $2) + FROM v_feed v WHERE v.id = $1", + ) + .bind(upload.id) + .bind(upload.user_id) + .fetch_optional(&state.pool) + .await + .ok() + .flatten(); + let (like_count, comment_count, liked_by_me) = counts.unwrap_or((0, 0, false)); + + UploadDto { + id: upload.id, + user_id: upload.user_id, + uploader_name: uploader_name.to_string(), + preview_url: upload + .preview_path + .as_ref() + .map(|_| format!("/api/v1/upload/{}/preview", upload.id)), + thumbnail_url: upload + .thumbnail_path + .as_ref() + .map(|_| format!("/api/v1/upload/{}/thumbnail", upload.id)), + mime_type: upload.mime_type.clone(), + caption: upload.caption.clone(), + hashtags, + like_count, + comment_count, + liked_by_me, + created_at: upload.created_at, + } +} + /// Drain a multipart body so the HTTP connection stays clean when returning an early error. /// Without draining, the client may still be sending the body after we've sent our response, /// which can corrupt the keep-alive connection for subsequent requests. diff --git a/backend/src/models/upload.rs b/backend/src/models/upload.rs index d27fbd7..2598930 100644 --- a/backend/src/models/upload.rs +++ b/backend/src/models/upload.rs @@ -53,6 +53,9 @@ pub struct VisibleMedia { impl Upload { /// Takes any executor so the caller can run it inside a transaction (atomic /// quota + insert) or standalone against the pool. + // Eight arguments, one per column the INSERT writes, with exactly one call site. A params + // struct here would restate the column list a second time and buy nothing. + #[allow(clippy::too_many_arguments)] pub async fn create<'e, E>( executor: E, event_id: Uuid, @@ -61,13 +64,23 @@ impl Upload { mime_type: &str, original_size_bytes: i64, caption: Option<&str>, - ) -> Result + client_upload_id: Option, + ) -> Result, sqlx::Error> where E: sqlx::PgExecutor<'e>, { + // `Ok(None)` means this exact `client_upload_id` is already stored — the caller's request + // is a retry of one that already succeeded, and it must replay the original row rather + // than create a second. Letting the unique index raise instead would work, but only after + // the whole transaction had aborted, and it would arrive as an opaque database error the + // caller would have to string-match to recognise. + // + // The conflict target repeats the index's `WHERE` clause because it is a partial index; + // without it Postgres cannot prove which index to use and rejects the statement. sqlx::query_as::<_, Self>( - "INSERT INTO upload (event_id, user_id, original_path, mime_type, original_size_bytes, caption) - VALUES ($1, $2, $3, $4, $5, $6) + "INSERT INTO upload (event_id, user_id, original_path, mime_type, original_size_bytes, caption, client_upload_id) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (client_upload_id) WHERE client_upload_id IS NOT NULL DO NOTHING RETURNING *", ) .bind(event_id) @@ -76,7 +89,29 @@ impl Upload { .bind(mime_type) .bind(original_size_bytes) .bind(caption) - .fetch_one(executor) + .bind(client_upload_id) + .fetch_optional(executor) + .await + } + + /// Look up a live upload by the idempotency key its client sent. + /// + /// Scoped to the user as well as the key: the key alone is unique, but a lookup that ignored + /// ownership would let one guest's retry return another guest's row if a key ever repeated. + /// Soft-deleted rows are excluded on purpose — if the guest deleted the photo and their queue + /// later retries, they should get a fresh upload rather than a resurrection of a deleted one. + pub async fn find_by_client_upload_id( + pool: &sqlx::PgPool, + user_id: Uuid, + client_upload_id: Uuid, + ) -> Result, sqlx::Error> { + sqlx::query_as::<_, Self>( + "SELECT * FROM upload + WHERE client_upload_id = $1 AND user_id = $2 AND deleted_at IS NULL", + ) + .bind(client_upload_id) + .bind(user_id) + .fetch_optional(pool) .await } diff --git a/backend/tests/upload_idempotency.rs b/backend/tests/upload_idempotency.rs new file mode 100644 index 0000000..0c55d84 --- /dev/null +++ b/backend/tests/upload_idempotency.rs @@ -0,0 +1,175 @@ +//! DB-backed tests for the upload idempotency key (migration 022). +//! +//! The guarantee under test is the one thing standing between a lost response and a duplicated +//! wedding photo: a retry of an upload that already committed must NOT create a second row, and +//! must not charge the guest's storage quota twice. The whole mechanism is SQL — a partial unique +//! index plus `ON CONFLICT DO NOTHING` — so it is tested against a real database with the real +//! migrations applied, using the same statements `src/` runs. + +mod common; + +use common::*; +use sqlx::PgPool; +use uuid::Uuid; + +/// SRC: `models/upload.rs::Upload::create` — the insert, verbatim. +/// +/// Returns the new row's id, or `None` when the key was already stored. The handler treats +/// `None` as "a concurrent retry won" and replays the stored row instead of committing. +async fn create_upload( + pool: &PgPool, + event_id: Uuid, + user_id: Uuid, + original_path: &str, + client_upload_id: Option, +) -> Option { + let row: Option<(Uuid,)> = sqlx::query_as( + "INSERT INTO upload (event_id, user_id, original_path, mime_type, original_size_bytes, caption, client_upload_id) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (client_upload_id) WHERE client_upload_id IS NOT NULL DO NOTHING + RETURNING id", + ) + .bind(event_id) + .bind(user_id) + .bind(original_path) + .bind("image/jpeg") + .bind(1_000i64) + .bind(Option::::None) + .bind(client_upload_id) + .fetch_optional(pool) + .await + .expect("create_upload"); + row.map(|(id,)| id) +} + +/// SRC: `models/upload.rs::Upload::find_by_client_upload_id` — the lookup, verbatim. +async fn find_by_key(pool: &PgPool, user_id: Uuid, client_upload_id: Uuid) -> Option { + let row: Option<(Uuid,)> = sqlx::query_as( + "SELECT id FROM upload + WHERE client_upload_id = $1 AND user_id = $2 AND deleted_at IS NULL", + ) + .bind(client_upload_id) + .bind(user_id) + .fetch_optional(pool) + .await + .expect("find_by_key"); + row.map(|(id,)| id) +} + +async fn upload_count(pool: &PgPool) -> i64 { + sqlx::query_scalar("SELECT COUNT(*) FROM upload") + .fetch_one(pool) + .await + .expect("upload_count") +} + +/// The core guarantee. A phone that loses the response and re-sends the same photo gets the +/// original row back, not a second copy in the gallery and a second charge against its quota. +#[sqlx::test] +async fn the_same_key_can_only_ever_store_one_upload(pool: PgPool) { + let event_id = seed_event(&pool, "wedding").await; + let user_id = seed_user(&pool, event_id, "Wackelige Wanda").await; + let key = Uuid::new_v4(); + + let first = create_upload(&pool, event_id, user_id, "originals/a.jpg", Some(key)).await; + assert!(first.is_some(), "the first attempt must store the upload"); + + // The retry: same key, and (as after a real re-send) a different file on disk. + let second = create_upload(&pool, event_id, user_id, "originals/b.jpg", Some(key)).await; + assert!( + second.is_none(), + "a retry of a committed upload must not insert a second row" + ); + assert_eq!(upload_count(&pool).await, 1); + + // And the handler can find the winner to replay it. + assert_eq!(find_by_key(&pool, user_id, key).await, first); +} + +/// The index must not over-reach. Two genuinely different photos carry different keys and must +/// both land — this is the ordinary case, and breaking it would silently drop uploads. +#[sqlx::test] +async fn different_keys_are_different_uploads(pool: PgPool) { + let event_id = seed_event(&pool, "wedding").await; + let user_id = seed_user(&pool, event_id, "Fleißige Frieda").await; + + for _ in 0..5 { + assert!( + create_upload( + &pool, + event_id, + user_id, + "originals/x.jpg", + Some(Uuid::new_v4()) + ) + .await + .is_some() + ); + } + assert_eq!(upload_count(&pool).await, 5); +} + +/// The index is PARTIAL, and this is why. Every upload that predates migration 022, and any +/// client that doesn't send a key, carries NULL — if those collided, the first such upload would +/// block every subsequent one and the whole event would fail after one photo. +#[sqlx::test] +async fn uploads_without_a_key_never_collide(pool: PgPool) { + let event_id = seed_event(&pool, "wedding").await; + let user_id = seed_user(&pool, event_id, "Alte Anna").await; + + for _ in 0..5 { + assert!( + create_upload(&pool, event_id, user_id, "originals/legacy.jpg", None) + .await + .is_some(), + "a NULL key must never be treated as a duplicate" + ); + } + assert_eq!(upload_count(&pool).await, 5); +} + +/// The lookup is scoped to the owner. The key alone is unique, so this can only matter if a key +/// ever repeated across users — but a replay that handed one guest another guest's upload row +/// would be a data leak, so the scope is asserted rather than assumed. +#[sqlx::test] +async fn the_replay_lookup_never_crosses_users(pool: PgPool) { + let event_id = seed_event(&pool, "wedding").await; + let owner = seed_user(&pool, event_id, "Besitzerin Bea").await; + let other = seed_user(&pool, event_id, "Fremder Franz").await; + let key = Uuid::new_v4(); + + let id = create_upload(&pool, event_id, owner, "originals/a.jpg", Some(key)).await; + + assert_eq!(find_by_key(&pool, owner, key).await, id); + assert_eq!( + find_by_key(&pool, other, key).await, + None, + "another guest's retry must not resolve to this upload" + ); +} + +/// A deleted photo must not be resurrected by a stale queue item. If the guest uploaded, deleted, +/// and their queue then retried the original request, replaying the deleted row would put the +/// photo they removed back in the gallery — so the lookup excludes soft-deleted rows and the +/// retry becomes a fresh upload instead. +#[sqlx::test] +async fn a_deleted_upload_is_not_replayed(pool: PgPool) { + let event_id = seed_event(&pool, "wedding").await; + let user_id = seed_user(&pool, event_id, "Reumütige Rita").await; + let key = Uuid::new_v4(); + + let id = create_upload(&pool, event_id, user_id, "originals/a.jpg", Some(key)) + .await + .expect("first insert"); + sqlx::query("UPDATE upload SET deleted_at = NOW() WHERE id = $1") + .bind(id) + .execute(&pool) + .await + .expect("soft delete"); + + assert_eq!( + find_by_key(&pool, user_id, key).await, + None, + "a soft-deleted upload must not be replayed" + ); +}