EventSnap/backend/tests/upload_concurrency.rs

//! DB-backed integration tests for the two concurrency guards in the upload commit path
//! (`handlers/upload.rs`). Both are SQL — a `FOR SHARE` row lock and an atomic compare-and-increment
//! — and both are load-bearing for things a user can actually lose: a wedding photo, or the disk.
//!
//! `#[sqlx::test]` gives each test a fresh database with the real migrations applied.

mod common;

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use chrono::{DateTime, Utc};
use common::*;
use sqlx::PgPool;
use uuid::Uuid;

/// SRC: `handlers/upload.rs:313-322` — the guarded quota increment, verbatim.
/// Returns `rows_affected()`; the handler aborts the whole upload tx when this is 0.
async fn quota_inc(exec: impl sqlx::PgExecutor<'_>, user_id: Uuid, size: i64, limit: i64) -> u64 {
    sqlx::query(
        "UPDATE \"user\" SET total_upload_bytes = total_upload_bytes + $2
         WHERE id = $1 AND total_upload_bytes + $2 <= $3",
    )
    .bind(user_id)
    .bind(size)
    .bind(limit)
    .execute(exec)
    .await
    .expect("quota_inc")
    .rows_affected()
}

async fn total_bytes(pool: &PgPool, user_id: Uuid) -> i64 {
    sqlx::query_scalar("SELECT total_upload_bytes FROM \"user\" WHERE id = $1")
        .bind(user_id)
        .fetch_one(pool)
        .await
        .expect("total_bytes")
}

// ─────────────────────────────────────────────────────────────────────────────
// 5. The atomic quota increment
// ─────────────────────────────────────────────────────────────────────────────

/// Two attempts sized off ONE stale snapshot, each of which "fits" on its own, cannot both commit.
/// The predicate re-reads `total_upload_bytes` inside the UPDATE, so the second matches 0 rows.
///
/// PREVENTS: one guest filling the disk. The handler's pre-flight quota check runs BEFORE the body is
/// streamed — minutes earlier, for a 500 MB video. If the commit trusted that snapshot, a guest could
/// start N uploads that each individually fit under the limit and land all N, blowing straight through
/// the quota and (in a 1 GB container) taking the event down for everyone.
#[sqlx::test]
async fn quota_two_attempts_from_one_stale_snapshot_cannot_both_commit(pool: PgPool) {
    let event_id = seed_event(&pool, "wedding").await;
    let user_id = seed_user(&pool, event_id, "Gierige Gudrun").await;

    const LIMIT: i64 = 100;
    const SIZE: i64 = 60;

    // THE STALE SNAPSHOT: the pre-flight check both uploads were admitted on.
    let snapshot = total_bytes(&pool, user_id).await;
    assert_eq!(snapshot, 0);
    // Each upload, judged against that snapshot alone, fits: 0 + 60 <= 100. Twice.
    assert!(snapshot + SIZE <= LIMIT);

    assert_eq!(
        quota_inc(&pool, user_id, SIZE, LIMIT).await,
        1,
        "the first upload commits"
    );
    assert_eq!(
        quota_inc(&pool, user_id, SIZE, LIMIT).await,
        0,
        "the second MUST affect 0 rows — it was admitted on a snapshot that is now a lie \
         (60 + 60 = 120 > 100). rows_affected() == 0 is what makes the handler abort."
    );

    assert_eq!(
        total_bytes(&pool, user_id).await,
        SIZE,
        "never 120 — the quota held"
    );
}

/// The same, but genuinely CONCURRENT: two transactions that both read `total = 0`, then both try to
/// commit 60 bytes against a 100-byte limit. The second UPDATE blocks on the first's row lock and —
/// because every predicate is on the ROW BEING UPDATED — Postgres re-evaluates it against the
/// post-commit row (EPQ) rather than the statement's original snapshot. It matches nothing.
///
/// PREVENTS: exactly the same disk-filling overrun, on the path it actually happens — two uploads
/// in flight at once, which is the normal case at a party.
#[sqlx::test]
async fn quota_guard_is_atomic_under_concurrent_transactions(pool: PgPool) {
    let event_id = seed_event(&pool, "wedding").await;
    let user_id = seed_user(&pool, event_id, "Gierige Gudrun").await;

    const LIMIT: i64 = 100;
    const SIZE: i64 = 60;

    let mut tx1 = pool.begin().await.unwrap();
    let mut tx2 = pool.begin().await.unwrap();

    // Both transactions read the same snapshot and both would pass a naive `total + size <= limit`
    // check done in Rust.
    for tx in [&mut tx1, &mut tx2] {
        let seen: i64 = sqlx::query_scalar("SELECT total_upload_bytes FROM \"user\" WHERE id = $1")
            .bind(user_id)
            .fetch_one(&mut **tx)
            .await
            .unwrap();
        assert_eq!(seen, 0, "both see an empty quota");
    }

    // tx1 takes the row lock and commits.
    assert_eq!(quota_inc(&mut *tx1, user_id, SIZE, LIMIT).await, 1);
    tx1.commit().await.unwrap();

    // tx2's UPDATE was written against the stale snapshot but is evaluated against the row as it
    // now stands.
    assert_eq!(
        quota_inc(&mut *tx2, user_id, SIZE, LIMIT).await,
        0,
        "the loser MUST see 0 rows affected — this is the entire quota guarantee"
    );
    tx2.rollback().await.unwrap();

    assert_eq!(total_bytes(&pool, user_id).await, SIZE);

    // And an upload that legitimately fits in what's left still succeeds — the guard rejects
    // overruns, not everything.
    assert_eq!(
        quota_inc(&pool, user_id, 40, LIMIT).await,
        1,
        "0 + 60 + 40 == 100, exactly at the limit"
    );
    assert_eq!(total_bytes(&pool, user_id).await, LIMIT);
    assert_eq!(
        quota_inc(&pool, user_id, 1, LIMIT).await,
        0,
        "and one byte more is refused"
    );
}

// ─────────────────────────────────────────────────────────────────────────────
// 6. The `FOR SHARE` upload lock vs. the release
// ─────────────────────────────────────────────────────────────────────────────

/// SRC: `handlers/upload.rs:297-303` — the in-transaction re-check under a row lock, verbatim.
async fn lock_and_read_event(
    tx: &mut sqlx::PgConnection,
    event_id: Uuid,
) -> (Option<DateTime<Utc>>, Option<DateTime<Utc>>) {
    sqlx::query_as(
        "SELECT uploads_locked_at, export_released_at FROM event WHERE id = $1 FOR SHARE",
    )
    .bind(event_id)
    .fetch_one(tx)
    .await
    .expect("FOR SHARE re-check")
}

/// THE GUARD AGAINST SILENT, PERMANENT PHOTO LOSS.
///
/// An upload holding `FOR SHARE` on the event row must BLOCK the `UPDATE event SET
/// export_released_at = NOW()` in `release_gallery` until it commits. Either the upload commits first
/// — and the release (hence the export snapshot) is strictly ordered after it, so the keepsake
/// CONTAINS the photo — or the release commits first and the upload observes the lock and rejects
/// (reversibly: the client keeps the blob and resumes after a reopen).
///
/// PREVENTS: the lost wedding photo. Without this serialization: a guest starts a 500 MB video, the
/// pre-flight lock check passes, the host releases the gallery, the export workers snapshot the
/// uploads table, and THEN the upload commits. The photo appears in the live feed but is missing from
/// the downloaded keepsake, forever — nothing ever regenerates it and nobody ever notices.
#[sqlx::test]
async fn for_share_upload_lock_serializes_against_release(pool: PgPool) {
    let event_id = seed_event(&pool, "wedding").await;
    let user_id = seed_user(&pool, event_id, "Fotograf Fritz").await;

    // ── The guest's upload transaction takes the share lock. ──
    let mut upload_tx = pool.begin().await.unwrap();
    let (locked, released) = lock_and_read_event(&mut upload_tx, event_id).await;
    assert!(
        locked.is_none() && released.is_none(),
        "uploads are open, so we proceed to commit"
    );

    // ── Concurrently, the host hits "Galerie freigeben". ──
    let release_done = Arc::new(AtomicBool::new(false));
    let release_task = {
        let pool = pool.clone();
        let release_done = release_done.clone();
        tokio::spawn(async move {
            sqlx::query(
                "UPDATE event
                 SET export_released_at = NOW(),
                     uploads_locked_at  = COALESCE(uploads_locked_at, NOW()),
                     export_epoch       = export_epoch + 1
                 WHERE id = $1 AND export_released_at IS NULL",
            )
            .bind(event_id)
            .execute(&pool)
            .await
            .expect("release");
            release_done.store(true, Ordering::SeqCst);
        })
    };

    // The release MUST be stuck behind our `FOR SHARE` row lock. (`FOR SHARE` conflicts with the
    // `FOR UPDATE` lock the UPDATE needs, so Postgres makes it wait — this is not a timing race,
    // it is a lock-conflict guarantee; the sleep only gives it every chance to wrongly proceed.)
    tokio::time::sleep(Duration::from_millis(750)).await;
    assert!(
        !release_done.load(Ordering::SeqCst),
        "the release MUST block while an upload holds FOR SHARE — if it can slip past, the export \
         snapshot is taken while a photo is still committing and that photo is lost forever"
    );

    // The photo commits. It is now unambiguously part of the upload set.
    let upload_id: Uuid = sqlx::query_scalar(
        "INSERT INTO upload (event_id, user_id, original_path, mime_type, original_size_bytes)
         VALUES ($1, $2, 'originals/wedding/x.jpg', 'image/jpeg', 1234) RETURNING id",
    )
    .bind(event_id)
    .bind(user_id)
    .fetch_one(&mut *upload_tx)
    .await
    .unwrap();
    upload_tx.commit().await.unwrap();

    // Only now can the release proceed.
    tokio::time::timeout(Duration::from_secs(5), release_task)
        .await
        .expect("the release must unblock once the upload commits")
        .unwrap();

    // THE PAYOFF: the export snapshot — the very query the ZIP worker runs — sees the photo. Order
    // enforced by the lock: upload commit < release < snapshot.
    let snapshot: Vec<Uuid> = sqlx::query_scalar(
        "SELECT u.id FROM upload u
           JOIN \"user\" usr ON usr.id = u.user_id
          WHERE u.event_id = $1 AND u.deleted_at IS NULL
            AND usr.uploads_hidden = FALSE AND usr.is_banned = FALSE",
    )
    .bind(event_id)
    .fetch_all(&pool)
    .await
    .unwrap();
    assert_eq!(
        snapshot,
        vec![upload_id],
        "the released keepsake CONTAINS the in-flight photo"
    );
}

/// The other side of the same lock: once the release has COMMITTED, the next upload's `FOR SHARE`
/// re-read sees `export_released_at` set and the handler rejects it with `UploadsLocked`.
///
/// PREVENTS: the same lost photo, on the losing side of the race — a photo committing AFTER the
/// export snapshot would be in the live feed but missing from the keepsake. Rejecting is the correct
/// outcome, and it is reversible: `UploadsLocked` (not Forbidden) tells the client to keep the blob
/// and resume when the host reopens.
#[sqlx::test]
async fn upload_after_release_commits_sees_the_lock_and_is_rejected(pool: PgPool) {
    let event_id = seed_event(&pool, "wedding").await;

    // Before the release, the re-check passes.
    let mut tx = pool.begin().await.unwrap();
    let (locked, released) = lock_and_read_event(&mut tx, event_id).await;
    assert!(locked.is_none() && released.is_none());
    tx.rollback().await.unwrap();

    assert_eq!(release_gallery(&pool, "wedding").await, Some(1));

    // After it, the identical re-check sees the release and the handler bails out.
    let mut tx = pool.begin().await.unwrap();
    let (locked, released) = lock_and_read_event(&mut tx, event_id).await;
    assert!(
        released.is_some(),
        "the FOR SHARE re-read MUST observe the committed release"
    );
    assert!(
        locked.is_some(),
        "release locks uploads in the same statement (release ⇒ lock)"
    );
    tx.rollback().await.unwrap();

    // And a reopen makes it uploadable again — the rejection was reversible, not terminal.
    assert_eq!(open_event(&pool, "wedding").await, 1);
    let mut tx = pool.begin().await.unwrap();
    let (locked, released) = lock_and_read_event(&mut tx, event_id).await;
    assert!(
        locked.is_none() && released.is_none(),
        "the guest can resume their upload"
    );
    tx.rollback().await.unwrap();
}