test(backend): DB-backed tests for the risky SQL; test isolation; clippy cleanup
All 40 backend tests were pure-function tests — not one line of SQL ran under `cargo test`,
even though the riskiest code in the repo is SQL. Add 12 DB-backed `#[sqlx::test]` tests
(fresh throwaway DB per test, real migrations) pinning the invariants that code is
load-bearing for, each mutation-verified to fail on broken code:
export_epoch.rs (8): epoch is post-increment on release (a worker born pre-increment is
inert); epoch strictly monotonic across reopen; a retired-epoch worker's writes are
no-ops; export_current is EXACTLY the invariant (8-case table); the ViewerOnly
carry-forward carries a done ZIP forward AND matches nothing when the ZIP is unfinished
(the af997a8 strand bug); boot recovery doesn't clobber a live ZIP.
upload_concurrency.rs (4): the atomic quota UPDATE stops two stale-snapshot uploads from
both committing (sequential AND concurrent-tx via EPQ); the FOR SHARE upload lock
serializes against release, blocking a photo from committing after the export snapshot.
Deleting FOR SHARE, or the carry-forward's `status='done'`, each makes a test fail.
Test isolation: TRUNCATE now also resets disk_cache and sse_tickets. The stale disk
reading was harmless only while quotas were globally off in e2e (they no longer are — see
the new quota spec), i.e. two holes were masking each other.
clippy: `cargo clippy --all-targets -- -D warnings` now passes (it did not). Deleted the
dead jobs.rs (an unused BackgroundJob sketch) and unused model methods; `#[allow(dead_code)]`
+ comment on the sqlx FromRow field-sets that ARE populated by the DB. No SQL or logic
changed. `cargo test` requires DATABASE_URL by design.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -75,10 +75,10 @@ impl IntoResponse for AppError {
|
|||||||
}
|
}
|
||||||
|
|
||||||
let mut resp = (status, axum::Json(body)).into_response();
|
let mut resp = (status, axum::Json(body)).into_response();
|
||||||
if let Some(secs) = retry_after_secs {
|
if let Some(secs) = retry_after_secs
|
||||||
if let Ok(val) = axum::http::HeaderValue::from_str(&secs.to_string()) {
|
&& let Ok(val) = axum::http::HeaderValue::from_str(&secs.to_string())
|
||||||
resp.headers_mut().insert(axum::http::header::RETRY_AFTER, val);
|
{
|
||||||
}
|
resp.headers_mut().insert(axum::http::header::RETRY_AFTER, val);
|
||||||
}
|
}
|
||||||
resp
|
resp
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -98,6 +98,10 @@ pub async fn get_config(
|
|||||||
Ok(Json(rows.into_iter().collect()))
|
Ok(Json(rows.into_iter().collect()))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Documents the wire shape of `PATCH /admin/config` (a flat `{key: value}` object).
|
||||||
|
/// `patch_config` extracts the `HashMap` directly rather than going through this newtype, so it is
|
||||||
|
/// never constructed in Rust — it stays as the serde-derived description of the request body.
|
||||||
|
#[allow(dead_code)]
|
||||||
#[derive(Deserialize)]
|
#[derive(Deserialize)]
|
||||||
pub struct PatchConfigRequest(pub HashMap<String, String>);
|
pub struct PatchConfigRequest(pub HashMap<String, String>);
|
||||||
|
|
||||||
|
|||||||
@@ -85,6 +85,19 @@ pub async fn truncate_all(
|
|||||||
// could serve the previous test's toggles.
|
// could serve the previous test's toggles.
|
||||||
state.config_cache.invalidate();
|
state.config_cache.invalidate();
|
||||||
|
|
||||||
|
// The other two in-memory singletons that TRUNCATE used to leave standing.
|
||||||
|
//
|
||||||
|
// `disk_cache` holds a free-space reading for up to its TTL. TRUNCATE has just deleted every
|
||||||
|
// uploaded file, which materially changes free space — so without this the next test can
|
||||||
|
// compute a storage quota from the PREVIOUS test's disk. That was harmless only while quotas
|
||||||
|
// were globally disabled in e2e (they no longer are: see specs/02-upload/quota.spec.ts, which
|
||||||
|
// steers the per-user limit off `free_disk_bytes`), i.e. two holes were masking each other.
|
||||||
|
state.disk_cache.invalidate();
|
||||||
|
|
||||||
|
// `sse_tickets` maps a ticket to a session token hash. TRUNCATE deletes the sessions, so every
|
||||||
|
// surviving ticket is a dangling reference to a user that no longer exists.
|
||||||
|
state.sse_tickets.clear();
|
||||||
|
|
||||||
Ok(StatusCode::NO_CONTENT)
|
Ok(StatusCode::NO_CONTENT)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -168,14 +168,14 @@ pub async fn upload(
|
|||||||
// Validate caption length. Counted in chars (code points) to match the
|
// Validate caption length. Counted in chars (code points) to match the
|
||||||
// "Zeichen" wording in the error message — `.len()` would be bytes and
|
// "Zeichen" wording in the error message — `.len()` would be bytes and
|
||||||
// reject perfectly valid German/emoji captions early.
|
// reject perfectly valid German/emoji captions early.
|
||||||
if let Some(ref cap) = caption {
|
if let Some(ref cap) = caption
|
||||||
if cap.chars().count() > MAX_CAPTION_LENGTH {
|
&& cap.chars().count() > MAX_CAPTION_LENGTH
|
||||||
let _ = tokio::fs::remove_file(&temp_abs).await;
|
{
|
||||||
return Err(AppError::BadRequest(format!(
|
let _ = tokio::fs::remove_file(&temp_abs).await;
|
||||||
"Beschreibung ist zu lang. Maximum: {} Zeichen.",
|
return Err(AppError::BadRequest(format!(
|
||||||
MAX_CAPTION_LENGTH
|
"Beschreibung ist zu lang. Maximum: {} Zeichen.",
|
||||||
)));
|
MAX_CAPTION_LENGTH
|
||||||
}
|
)));
|
||||||
}
|
}
|
||||||
|
|
||||||
// Determine the file type from its magic bytes and require it to be on the
|
// Determine the file type from its magic bytes and require it to be on the
|
||||||
@@ -555,6 +555,9 @@ pub struct QuotaEstimate {
|
|||||||
pub limit_bytes: Option<i64>,
|
pub limit_bytes: Option<i64>,
|
||||||
pub active_uploaders: i64,
|
pub active_uploaders: i64,
|
||||||
pub free_disk_bytes: i64,
|
pub free_disk_bytes: i64,
|
||||||
|
/// The tolerance factor the limit above was computed with. Carried on the snapshot so the
|
||||||
|
/// number is self-describing; no caller reads it back today.
|
||||||
|
#[allow(dead_code)]
|
||||||
pub tolerance: f64,
|
pub tolerance: f64,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -3,6 +3,10 @@ use serde::Serialize;
|
|||||||
use sqlx::PgPool;
|
use sqlx::PgPool;
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
|
// Row shape for `comment`: every field is populated by sqlx from `SELECT *` / `RETURNING *`.
|
||||||
|
// `deleted_at` is not read in Rust today (the soft-delete filter lives in SQL), but it is part of
|
||||||
|
// the row and stays here so the struct keeps mirroring the table.
|
||||||
|
#[allow(dead_code)]
|
||||||
#[derive(Debug, sqlx::FromRow)]
|
#[derive(Debug, sqlx::FromRow)]
|
||||||
pub struct Comment {
|
pub struct Comment {
|
||||||
pub id: Uuid,
|
pub id: Uuid,
|
||||||
@@ -87,16 +91,8 @@ impl Comment {
|
|||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn soft_delete(pool: &PgPool, id: Uuid) -> Result<(), sqlx::Error> {
|
/// Event-scoped soft delete. Returns `false` if the comment doesn't exist or belongs to a
|
||||||
sqlx::query("UPDATE comment SET deleted_at = NOW() WHERE id = $1")
|
/// different event.
|
||||||
.bind(id)
|
|
||||||
.execute(pool)
|
|
||||||
.await?;
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Event-scoped variant of [`Self::soft_delete`]. Returns `false` if the
|
|
||||||
/// comment doesn't exist or belongs to a different event.
|
|
||||||
/// Executor-generic so the delete and the keepsake regeneration can share one transaction
|
/// Executor-generic so the delete and the keepsake regeneration can share one transaction
|
||||||
/// (see `Upload::soft_delete_in_event` for why that must be atomic).
|
/// (see `Upload::soft_delete_in_event` for why that must be atomic).
|
||||||
pub async fn soft_delete_in_event(
|
pub async fn soft_delete_in_event(
|
||||||
|
|||||||
@@ -2,6 +2,11 @@ use chrono::{DateTime, Utc};
|
|||||||
use sqlx::PgPool;
|
use sqlx::PgPool;
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
|
// Row shape for `event`: every field is populated by sqlx from `SELECT *` / `RETURNING *`. Several
|
||||||
|
// (`slug`, `cover_image_path`, `export_epoch`, `created_at`) are not read through this struct today
|
||||||
|
// — callers that need them query the column directly — but they are part of the row and stay here so
|
||||||
|
// the struct keeps mirroring the table.
|
||||||
|
#[allow(dead_code)]
|
||||||
#[derive(Debug, sqlx::FromRow)]
|
#[derive(Debug, sqlx::FromRow)]
|
||||||
pub struct Event {
|
pub struct Event {
|
||||||
pub id: Uuid,
|
pub id: Uuid,
|
||||||
|
|||||||
@@ -1,6 +1,8 @@
|
|||||||
use sqlx::PgPool;
|
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
|
// Row shape for `hashtag`, populated by sqlx from `RETURNING *` in `upsert`. Callers only use
|
||||||
|
// `id` today; `event_id`/`tag` are the rest of the row and stay part of the struct.
|
||||||
|
#[allow(dead_code)]
|
||||||
#[derive(Debug, sqlx::FromRow)]
|
#[derive(Debug, sqlx::FromRow)]
|
||||||
pub struct Hashtag {
|
pub struct Hashtag {
|
||||||
pub id: Uuid,
|
pub id: Uuid,
|
||||||
@@ -61,22 +63,6 @@ impl Hashtag {
|
|||||||
.await?;
|
.await?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn tags_for_upload(
|
|
||||||
pool: &PgPool,
|
|
||||||
upload_id: Uuid,
|
|
||||||
) -> Result<Vec<String>, sqlx::Error> {
|
|
||||||
let rows: Vec<(String,)> = sqlx::query_as(
|
|
||||||
"SELECT h.tag FROM hashtag h
|
|
||||||
JOIN upload_hashtag uh ON uh.hashtag_id = h.id
|
|
||||||
WHERE uh.upload_id = $1
|
|
||||||
ORDER BY h.tag",
|
|
||||||
)
|
|
||||||
.bind(upload_id)
|
|
||||||
.fetch_all(pool)
|
|
||||||
.await?;
|
|
||||||
Ok(rows.into_iter().map(|r| r.0).collect())
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Extract `#hashtags` from text (caption or body). Tags are restricted to
|
/// Extract `#hashtags` from text (caption or body). Tags are restricted to
|
||||||
|
|||||||
@@ -2,6 +2,9 @@ use chrono::{DateTime, Utc};
|
|||||||
use sqlx::PgPool;
|
use sqlx::PgPool;
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
|
// Row shape for `session`, populated by sqlx from `RETURNING *`. Session validation is done in SQL
|
||||||
|
// (expiry/last-seen predicates), so no field is read in Rust — the struct is the row's shape.
|
||||||
|
#[allow(dead_code)]
|
||||||
#[derive(Debug, sqlx::FromRow)]
|
#[derive(Debug, sqlx::FromRow)]
|
||||||
pub struct Session {
|
pub struct Session {
|
||||||
pub id: Uuid,
|
pub id: Uuid,
|
||||||
|
|||||||
@@ -3,6 +3,10 @@ use serde::Serialize;
|
|||||||
use sqlx::PgPool;
|
use sqlx::PgPool;
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
|
// Row shape for `upload`: every field is populated by sqlx from `RETURNING *` in `create`. Callers
|
||||||
|
// mostly use `id` and hand the rest to the compression/feed queries, so most fields are never read
|
||||||
|
// through this struct — they stay here so it keeps mirroring the table.
|
||||||
|
#[allow(dead_code)]
|
||||||
#[derive(Debug, sqlx::FromRow)]
|
#[derive(Debug, sqlx::FromRow)]
|
||||||
pub struct Upload {
|
pub struct Upload {
|
||||||
pub id: Uuid,
|
pub id: Uuid,
|
||||||
@@ -75,15 +79,6 @@ impl Upload {
|
|||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn find_by_id(pool: &PgPool, id: Uuid) -> Result<Option<Self>, sqlx::Error> {
|
|
||||||
sqlx::query_as::<_, Self>(
|
|
||||||
"SELECT * FROM upload WHERE id = $1 AND deleted_at IS NULL",
|
|
||||||
)
|
|
||||||
.bind(id)
|
|
||||||
.fetch_optional(pool)
|
|
||||||
.await
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Lean lookup for the public media aliases (`get_original`/`get_preview`/
|
/// Lean lookup for the public media aliases (`get_original`/`get_preview`/
|
||||||
/// `get_thumbnail`): returns ONLY the file paths + mime for a visible upload —
|
/// `get_thumbnail`): returns ONLY the file paths + mime for a visible upload —
|
||||||
/// excluding soft-deleted rows, hidden owners (`uploads_hidden`), and banned owners
|
/// excluding soft-deleted rows, hidden owners (`uploads_hidden`), and banned owners
|
||||||
|
|||||||
@@ -22,6 +22,10 @@ impl UserRole {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Row shape for `user`: every field is populated by sqlx from `SELECT *` / `RETURNING *`.
|
||||||
|
// `uploads_hidden`, `failed_pin_attempts` and `created_at` are enforced/updated in SQL rather than
|
||||||
|
// read in Rust, but they are part of the row and stay here so the struct keeps mirroring the table.
|
||||||
|
#[allow(dead_code)]
|
||||||
#[derive(Debug, sqlx::FromRow)]
|
#[derive(Debug, sqlx::FromRow)]
|
||||||
pub struct User {
|
pub struct User {
|
||||||
pub id: Uuid,
|
pub id: Uuid,
|
||||||
|
|||||||
@@ -34,6 +34,17 @@ impl DiskCache {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Drop the cached reading so the next `snapshot()` re-measures the filesystem.
|
||||||
|
///
|
||||||
|
/// Used by the e2e TRUNCATE endpoint. Truncating deletes every uploaded file, which materially
|
||||||
|
/// changes free space — but the cached reading survives for up to the TTL, so the next test can
|
||||||
|
/// compute a quota from the PREVIOUS test's disk. That matters now that the quota tests steer
|
||||||
|
/// the per-user limit off `free_disk_bytes`: a stale reading makes the limit wrong and the test
|
||||||
|
/// flaky, for reasons that have nothing to do with the code under test.
|
||||||
|
pub fn invalidate(&self) {
|
||||||
|
*self.inner.write().unwrap() = None;
|
||||||
|
}
|
||||||
|
|
||||||
/// Cached `(total, free)` bytes for the filesystem that holds `media_path`.
|
/// Cached `(total, free)` bytes for the filesystem that holds `media_path`.
|
||||||
///
|
///
|
||||||
/// Returns `None` when the mount can't be resolved — callers MUST treat that as
|
/// Returns `None` when the mount can't be resolved — callers MUST treat that as
|
||||||
@@ -47,10 +58,11 @@ impl DiskCache {
|
|||||||
/// caller to ask about a different volume (e.g. the exports volume, which is a separate mount)
|
/// caller to ask about a different volume (e.g. the exports volume, which is a separate mount)
|
||||||
/// would have silently got the wrong filesystem's free space.
|
/// would have silently got the wrong filesystem's free space.
|
||||||
pub fn snapshot(&self, path: &Path) -> Option<DiskInfo> {
|
pub fn snapshot(&self, path: &Path) -> Option<DiskInfo> {
|
||||||
if let Some((cached_path, info, at)) = self.inner.read().unwrap().as_ref() {
|
if let Some((cached_path, info, at)) = self.inner.read().unwrap().as_ref()
|
||||||
if cached_path == path && at.elapsed() < TTL {
|
&& cached_path == path
|
||||||
return Some(*info);
|
&& at.elapsed() < TTL
|
||||||
}
|
{
|
||||||
|
return Some(*info);
|
||||||
}
|
}
|
||||||
let info = read_disk_for_path(path)?;
|
let info = read_disk_for_path(path)?;
|
||||||
*self.inner.write().unwrap() = Some((path.to_path_buf(), info, Instant::now()));
|
*self.inner.write().unwrap() = Some((path.to_path_buf(), info, Instant::now()));
|
||||||
|
|||||||
@@ -389,6 +389,10 @@ async fn invalidate_missing_files(
|
|||||||
/// start immediately.
|
/// start immediately.
|
||||||
pub const REGEN_DEBOUNCE: Duration = Duration::from_secs(20);
|
pub const REGEN_DEBOUNCE: Duration = Duration::from_secs(20);
|
||||||
|
|
||||||
|
// Export worker entry point: every argument is state the spawned worker is BORN with (notably
|
||||||
|
// `epoch`). Bundling them into a struct would be a pure-refactor risk on the epoch logic for no
|
||||||
|
// gain, so the arity stands.
|
||||||
|
#[allow(clippy::too_many_arguments)]
|
||||||
pub fn spawn_export_jobs(
|
pub fn spawn_export_jobs(
|
||||||
event_id: Uuid,
|
event_id: Uuid,
|
||||||
event_name: String,
|
event_name: String,
|
||||||
|
|||||||
@@ -2,7 +2,6 @@ pub mod compression;
|
|||||||
pub mod config;
|
pub mod config;
|
||||||
pub mod disk;
|
pub mod disk;
|
||||||
pub mod export;
|
pub mod export;
|
||||||
pub mod jobs;
|
|
||||||
pub mod maintenance;
|
pub mod maintenance;
|
||||||
pub mod rate_limiter;
|
pub mod rate_limiter;
|
||||||
pub mod sse_tickets;
|
pub mod sse_tickets;
|
||||||
|
|||||||
@@ -123,12 +123,58 @@ mod tests {
|
|||||||
assert!(rl.check("k", 1, w), "the slot should expire once the window passes");
|
assert!(rl.check("k", 1, w), "the slot should expire once the window passes");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// `retry_after` is not a "some number in range" — it is the time until the oldest slot
|
||||||
|
/// in the window frees up, and it is surfaced to clients as the backoff they sleep for
|
||||||
|
/// (see `upload-queue.ts`). Asserting only `(1..=60)` spans the entire reachable domain
|
||||||
|
/// of a 60s window, so a hardcoded `Err(1)` would satisfy it while telling every client
|
||||||
|
/// to hammer the server a second later. Pin the actual value.
|
||||||
#[test]
|
#[test]
|
||||||
fn retry_after_is_between_one_and_window() {
|
fn retry_after_is_the_remaining_window() {
|
||||||
let rl = RateLimiter::new();
|
let rl = RateLimiter::new();
|
||||||
assert!(rl.check_with_retry("k", 1, MIN).is_ok());
|
|
||||||
let retry = rl.check_with_retry("k", 1, MIN).unwrap_err();
|
// The slot was consumed just now, so essentially the whole window remains.
|
||||||
assert!((1..=60).contains(&retry), "retry_after {retry} out of range");
|
// `as_secs()` truncates the sub-second remainder, so a 30s window reports 29.
|
||||||
|
let w30 = Duration::from_secs(30);
|
||||||
|
assert!(rl.check_with_retry("a", 1, w30).is_ok());
|
||||||
|
let a = rl.check_with_retry("a", 1, w30).unwrap_err();
|
||||||
|
assert_eq!(a, 29, "retry_after must be the remaining window, got {a}");
|
||||||
|
|
||||||
|
// A different window must yield a different retry_after: no single constant can
|
||||||
|
// satisfy both this and the assertion above.
|
||||||
|
let w10 = Duration::from_secs(10);
|
||||||
|
assert!(rl.check_with_retry("b", 1, w10).is_ok());
|
||||||
|
let b = rl.check_with_retry("b", 1, w10).unwrap_err();
|
||||||
|
assert_eq!(b, 9, "retry_after must scale with the window, got {b}");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn retry_after_counts_down_as_the_window_elapses() {
|
||||||
|
let rl = RateLimiter::new();
|
||||||
|
let w = Duration::from_secs(30);
|
||||||
|
assert!(rl.check_with_retry("k", 1, w).is_ok());
|
||||||
|
let first = rl.check_with_retry("k", 1, w).unwrap_err();
|
||||||
|
|
||||||
|
std::thread::sleep(Duration::from_millis(1200));
|
||||||
|
let second = rl.check_with_retry("k", 1, w).unwrap_err();
|
||||||
|
|
||||||
|
// A client that waits 1.2s must be told to wait ~1.2s less — otherwise the advertised
|
||||||
|
// backoff is a constant, not a deadline.
|
||||||
|
let shaved = first - second;
|
||||||
|
assert!(
|
||||||
|
(1..=2).contains(&shaved),
|
||||||
|
"1.2s of waiting must shorten the advertised backoff by ~1s (got {first} then {second})"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn retry_after_floors_at_one_second() {
|
||||||
|
let rl = RateLimiter::new();
|
||||||
|
let w = Duration::from_millis(800);
|
||||||
|
assert!(rl.check_with_retry("k", 1, w).is_ok());
|
||||||
|
let retry = rl.check_with_retry("k", 1, w).unwrap_err();
|
||||||
|
// The sub-second remainder truncates to 0; clients must never be told "retry in 0s"
|
||||||
|
// (that's a busy-loop). The `.max(1)` floor is what prevents it.
|
||||||
|
assert_eq!(retry, 1, "a sub-second remainder must floor to 1, got {retry}");
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
@@ -140,6 +186,59 @@ mod tests {
|
|||||||
assert!(rl.check("k", 1, MIN), "clear() must free the window");
|
assert!(rl.check("k", 1, MIN), "clear() must free the window");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// `prune()` is a memory-leak guard: without it a long-lived process keeps one HashMap
|
||||||
|
/// entry per IP that ever connected. Nothing in the public API observes the map size, so
|
||||||
|
/// the only way to catch a no-op body (`fn prune(&self) {}`) is to look at the map — the
|
||||||
|
/// tests module can see the private field.
|
||||||
|
#[test]
|
||||||
|
fn prune_drops_keys_whose_windows_have_fully_expired() {
|
||||||
|
let rl = RateLimiter::new();
|
||||||
|
|
||||||
|
// A key whose only timestamp is older than the 24h ceiling. We can't sleep for a day,
|
||||||
|
// so backdate the Instant directly.
|
||||||
|
let ancient = Instant::now()
|
||||||
|
.checked_sub(Duration::from_secs(25 * 60 * 60))
|
||||||
|
.expect("backdating an Instant by 25h");
|
||||||
|
rl.windows
|
||||||
|
.lock()
|
||||||
|
.unwrap()
|
||||||
|
.insert("stale".to_string(), vec![ancient]);
|
||||||
|
|
||||||
|
// ...alongside a key that is still inside its window.
|
||||||
|
assert!(rl.check("live", 5, MIN));
|
||||||
|
assert_eq!(rl.windows.lock().unwrap().len(), 2);
|
||||||
|
|
||||||
|
rl.prune();
|
||||||
|
|
||||||
|
let map = rl.windows.lock().unwrap();
|
||||||
|
assert!(
|
||||||
|
!map.contains_key("stale"),
|
||||||
|
"prune() must drop keys whose timestamps have all expired"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
map.contains_key("live"),
|
||||||
|
"prune() must keep keys that still have live timestamps"
|
||||||
|
);
|
||||||
|
assert_eq!(map.len(), 1, "exactly one key should survive the prune");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn prune_does_not_reset_a_live_window() {
|
||||||
|
// The counterpart to the test above: pruning must reclaim memory, never quota. If
|
||||||
|
// prune() dropped live keys, every background sweep would hand attackers a fresh
|
||||||
|
// budget.
|
||||||
|
let rl = RateLimiter::new();
|
||||||
|
assert!(rl.check("k", 1, MIN));
|
||||||
|
assert!(!rl.check("k", 1, MIN));
|
||||||
|
|
||||||
|
rl.prune();
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
!rl.check("k", 1, MIN),
|
||||||
|
"prune() must not clear a window that is still active"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn client_ip_takes_rightmost_forwarded_for_entry() {
|
fn client_ip_takes_rightmost_forwarded_for_entry() {
|
||||||
// The right-most entry is the hop our trusted proxy (Caddy) appended.
|
// The right-most entry is the hop our trusted proxy (Caddy) appended.
|
||||||
|
|||||||
@@ -31,6 +31,13 @@ impl SseTicketStore {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Drop every outstanding ticket. Used by the e2e TRUNCATE endpoint: tickets are bound to a
|
||||||
|
/// session token hash, and TRUNCATE deletes the sessions out from under them, so anything left
|
||||||
|
/// here is a dangling reference to a user that no longer exists.
|
||||||
|
pub fn clear(&self) {
|
||||||
|
self.inner.lock().unwrap().clear();
|
||||||
|
}
|
||||||
|
|
||||||
/// Mint a new ticket bound to the caller's session (identified by token hash).
|
/// Mint a new ticket bound to the caller's session (identified by token hash).
|
||||||
pub fn issue(&self, token_hash: String) -> String {
|
pub fn issue(&self, token_hash: String) -> String {
|
||||||
let ticket = random_ticket();
|
let ticket = random_ticket();
|
||||||
|
|||||||
257
backend/tests/common/mod.rs
Normal file
257
backend/tests/common/mod.rs
Normal file
@@ -0,0 +1,257 @@
|
|||||||
|
//! Shared fixtures for the DB-backed integration tests.
|
||||||
|
//!
|
||||||
|
//! Every helper here executes SQL that is **character-for-character identical** to what `src/`
|
||||||
|
//! actually runs (see the `// SRC:` markers). That is the whole point: a paraphrased query is a
|
||||||
|
//! query nobody runs, and a test that passes against a paraphrase proves nothing about production.
|
||||||
|
|
||||||
|
#![allow(dead_code)] // each integration-test crate uses a different subset of these helpers
|
||||||
|
|
||||||
|
use sqlx::PgPool;
|
||||||
|
use uuid::Uuid;
|
||||||
|
|
||||||
|
/// Insert a bare, unreleased event (epoch 0, uploads open).
|
||||||
|
pub async fn seed_event(pool: &PgPool, slug: &str) -> Uuid {
|
||||||
|
sqlx::query_scalar("INSERT INTO event (slug, name) VALUES ($1, $2) RETURNING id")
|
||||||
|
.bind(slug)
|
||||||
|
.bind("Hochzeit")
|
||||||
|
.fetch_one(pool)
|
||||||
|
.await
|
||||||
|
.expect("seed event")
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Insert a guest with a zeroed byte total.
|
||||||
|
pub async fn seed_user(pool: &PgPool, event_id: Uuid, name: &str) -> Uuid {
|
||||||
|
sqlx::query_scalar(
|
||||||
|
"INSERT INTO \"user\" (event_id, display_name, recovery_pin_hash)
|
||||||
|
VALUES ($1, $2, 'x') RETURNING id",
|
||||||
|
)
|
||||||
|
.bind(event_id)
|
||||||
|
.bind(name)
|
||||||
|
.fetch_one(pool)
|
||||||
|
.await
|
||||||
|
.expect("seed user")
|
||||||
|
}
|
||||||
|
|
||||||
|
/// SRC: `handlers/host.rs::release_gallery` — the claim + epoch bump, verbatim.
|
||||||
|
/// Returns the POST-increment epoch, exactly as the handler consumes it.
|
||||||
|
pub async fn release_gallery(pool: &PgPool, slug: &str) -> Option<i64> {
|
||||||
|
let claimed: Option<(Uuid, String, i64)> = sqlx::query_as(
|
||||||
|
"UPDATE event
|
||||||
|
SET export_released_at = NOW(),
|
||||||
|
uploads_locked_at = COALESCE(uploads_locked_at, NOW()),
|
||||||
|
export_epoch = export_epoch + 1
|
||||||
|
WHERE slug = $1 AND export_released_at IS NULL
|
||||||
|
RETURNING id, name, export_epoch",
|
||||||
|
)
|
||||||
|
.bind(slug)
|
||||||
|
.fetch_optional(pool)
|
||||||
|
.await
|
||||||
|
.expect("release_gallery");
|
||||||
|
|
||||||
|
if let Some((event_id, _, epoch)) = claimed {
|
||||||
|
// The handler arms both jobs in the SAME transaction; for a single-connection fixture the
|
||||||
|
// sequencing is equivalent.
|
||||||
|
let mut conn = pool.acquire().await.expect("acquire");
|
||||||
|
enqueue_types_at_epoch(&mut conn, event_id, epoch, &["zip", "html"]).await;
|
||||||
|
Some(epoch)
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// SRC: `handlers/host.rs::open_event` — the one statement that retires an entire generation.
|
||||||
|
/// Returns rows affected.
|
||||||
|
pub async fn open_event(pool: &PgPool, slug: &str) -> u64 {
|
||||||
|
sqlx::query(
|
||||||
|
"UPDATE event
|
||||||
|
SET uploads_locked_at = NULL,
|
||||||
|
export_released_at = NULL,
|
||||||
|
export_epoch = export_epoch + 1
|
||||||
|
WHERE slug = $1 AND (uploads_locked_at IS NOT NULL OR export_released_at IS NOT NULL)",
|
||||||
|
)
|
||||||
|
.bind(slug)
|
||||||
|
.execute(pool)
|
||||||
|
.await
|
||||||
|
.expect("open_event")
|
||||||
|
.rows_affected()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// SRC: `services/export.rs::enqueue_types_at_epoch` — verbatim upsert.
|
||||||
|
pub async fn enqueue_types_at_epoch(
|
||||||
|
conn: &mut sqlx::PgConnection,
|
||||||
|
event_id: Uuid,
|
||||||
|
epoch: i64,
|
||||||
|
types: &[&str],
|
||||||
|
) {
|
||||||
|
for export_type in types {
|
||||||
|
sqlx::query(
|
||||||
|
"INSERT INTO export_job (event_id, type, status, progress_pct, epoch)
|
||||||
|
VALUES ($1, $2::export_type, 'pending', 0, $3)
|
||||||
|
ON CONFLICT (event_id, type) DO UPDATE
|
||||||
|
SET status = 'pending', progress_pct = 0, file_path = NULL,
|
||||||
|
error_message = NULL, completed_at = NULL,
|
||||||
|
epoch = EXCLUDED.epoch
|
||||||
|
WHERE export_job.status <> 'done' OR export_job.epoch <> EXCLUDED.epoch",
|
||||||
|
)
|
||||||
|
.bind(event_id)
|
||||||
|
.bind(export_type)
|
||||||
|
.bind(epoch)
|
||||||
|
.execute(&mut *conn)
|
||||||
|
.await
|
||||||
|
.expect("enqueue_types_at_epoch");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// SRC: `services/export.rs::claim_job` — verbatim. `true` = we won the generation.
|
||||||
|
pub async fn claim_job(pool: &PgPool, event_id: Uuid, export_type: &str, epoch: i64) -> bool {
|
||||||
|
sqlx::query(
|
||||||
|
"UPDATE export_job SET status = 'running'
|
||||||
|
WHERE event_id = $1 AND type = $2::export_type
|
||||||
|
AND epoch = $3 AND status = 'pending'",
|
||||||
|
)
|
||||||
|
.bind(event_id)
|
||||||
|
.bind(export_type)
|
||||||
|
.bind(epoch)
|
||||||
|
.execute(pool)
|
||||||
|
.await
|
||||||
|
.expect("claim_job")
|
||||||
|
.rows_affected()
|
||||||
|
> 0
|
||||||
|
}
|
||||||
|
|
||||||
|
/// SRC: `services/export.rs::finalize_job` — verbatim. This IS the publish step.
|
||||||
|
pub async fn finalize_job(
|
||||||
|
pool: &PgPool,
|
||||||
|
event_id: Uuid,
|
||||||
|
export_type: &str,
|
||||||
|
epoch: i64,
|
||||||
|
file_path: &str,
|
||||||
|
) -> bool {
|
||||||
|
sqlx::query(
|
||||||
|
"UPDATE export_job
|
||||||
|
SET status = 'done', progress_pct = 100, file_path = $3, completed_at = NOW()
|
||||||
|
WHERE event_id = $1 AND type = $2::export_type
|
||||||
|
AND epoch = $4 AND status = 'running'",
|
||||||
|
)
|
||||||
|
.bind(event_id)
|
||||||
|
.bind(export_type)
|
||||||
|
.bind(file_path)
|
||||||
|
.bind(epoch)
|
||||||
|
.execute(pool)
|
||||||
|
.await
|
||||||
|
.expect("finalize_job")
|
||||||
|
.rows_affected()
|
||||||
|
> 0
|
||||||
|
}
|
||||||
|
|
||||||
|
/// SRC: `services/export.rs::update_progress` — verbatim. Doubles as the worker's liveness check:
|
||||||
|
/// `false` means "your generation was retired, stop working".
|
||||||
|
pub async fn update_progress(
|
||||||
|
pool: &PgPool,
|
||||||
|
event_id: Uuid,
|
||||||
|
export_type: &str,
|
||||||
|
epoch: i64,
|
||||||
|
pct: i16,
|
||||||
|
) -> bool {
|
||||||
|
sqlx::query(
|
||||||
|
"UPDATE export_job SET progress_pct = $3
|
||||||
|
WHERE event_id = $1 AND type = $2::export_type
|
||||||
|
AND epoch = $4 AND status = 'running'",
|
||||||
|
)
|
||||||
|
.bind(event_id)
|
||||||
|
.bind(export_type)
|
||||||
|
.bind(pct)
|
||||||
|
.bind(epoch)
|
||||||
|
.execute(pool)
|
||||||
|
.await
|
||||||
|
.expect("update_progress")
|
||||||
|
.rows_affected()
|
||||||
|
> 0
|
||||||
|
}
|
||||||
|
|
||||||
|
/// SRC: `services/export.rs::invalidate_and_arm` — the ViewerOnly ZIP carry-forward, verbatim.
|
||||||
|
/// Returns `rows_affected() == 1`, which is what the production code branches on.
|
||||||
|
pub async fn carry_zip_forward(pool: &PgPool, event_id: Uuid, epoch: i64) -> bool {
|
||||||
|
sqlx::query(
|
||||||
|
"UPDATE export_job SET epoch = $2
|
||||||
|
WHERE event_id = $1 AND type = 'zip'::export_type
|
||||||
|
AND status = 'done' AND epoch = $2 - 1",
|
||||||
|
)
|
||||||
|
.bind(event_id)
|
||||||
|
.bind(epoch)
|
||||||
|
.execute(pool)
|
||||||
|
.await
|
||||||
|
.expect("carry_zip_forward")
|
||||||
|
.rows_affected()
|
||||||
|
== 1
|
||||||
|
}
|
||||||
|
|
||||||
|
/// SRC: `services/export.rs::invalidate_and_arm` — the epoch bump, verbatim.
|
||||||
|
pub async fn bump_epoch(pool: &PgPool, slug: &str) -> Option<(Uuid, String, i64)> {
|
||||||
|
sqlx::query_as(
|
||||||
|
"UPDATE event SET export_epoch = export_epoch + 1
|
||||||
|
WHERE slug = $1 AND export_released_at IS NOT NULL
|
||||||
|
RETURNING id, name, export_epoch",
|
||||||
|
)
|
||||||
|
.bind(slug)
|
||||||
|
.fetch_optional(pool)
|
||||||
|
.await
|
||||||
|
.expect("bump_epoch")
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The event's authoritative epoch.
|
||||||
|
pub async fn event_epoch(pool: &PgPool, event_id: Uuid) -> i64 {
|
||||||
|
sqlx::query_scalar("SELECT export_epoch FROM event WHERE id = $1")
|
||||||
|
.bind(event_id)
|
||||||
|
.fetch_one(pool)
|
||||||
|
.await
|
||||||
|
.expect("event_epoch")
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The raw job row, bypassing `export_current` — what the WORKER sees.
|
||||||
|
pub async fn job_row(
|
||||||
|
pool: &PgPool,
|
||||||
|
event_id: Uuid,
|
||||||
|
export_type: &str,
|
||||||
|
) -> Option<(String, i64, Option<String>)> {
|
||||||
|
sqlx::query_as(
|
||||||
|
"SELECT status::text, epoch, file_path FROM export_job
|
||||||
|
WHERE event_id = $1 AND type = $2::export_type",
|
||||||
|
)
|
||||||
|
.bind(event_id)
|
||||||
|
.bind(export_type)
|
||||||
|
.fetch_optional(pool)
|
||||||
|
.await
|
||||||
|
.expect("job_row")
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Does `export_current` expose this job at all? (The view itself, without the `status` filter —
|
||||||
|
/// it is what `handlers/admin.rs::export_status` reports to the host UI.)
|
||||||
|
pub async fn in_export_current(pool: &PgPool, event_id: Uuid, export_type: &str) -> bool {
|
||||||
|
sqlx::query_scalar::<_, i64>(
|
||||||
|
"SELECT COUNT(*) FROM export_current
|
||||||
|
WHERE event_id = $1 AND type = $2::export_type",
|
||||||
|
)
|
||||||
|
.bind(event_id)
|
||||||
|
.bind(export_type)
|
||||||
|
.fetch_one(pool)
|
||||||
|
.await
|
||||||
|
.expect("in_export_current")
|
||||||
|
> 0
|
||||||
|
}
|
||||||
|
|
||||||
|
/// THE download predicate. SRC: `handlers/admin.rs::download_export` reads exactly this shape —
|
||||||
|
/// `SELECT c.file_path FROM export_current c WHERE ... AND c.status = 'done'`. If this returns
|
||||||
|
/// `Some`, a guest can download the keepsake; if `None`, they get a 404.
|
||||||
|
pub async fn downloadable(pool: &PgPool, event_id: Uuid, export_type: &str) -> Option<String> {
|
||||||
|
sqlx::query_scalar(
|
||||||
|
"SELECT c.file_path FROM export_current c
|
||||||
|
WHERE c.event_id = $1 AND c.type = $2::export_type AND c.status = 'done'",
|
||||||
|
)
|
||||||
|
.bind(event_id)
|
||||||
|
.bind(export_type)
|
||||||
|
.fetch_optional(pool)
|
||||||
|
.await
|
||||||
|
.expect("downloadable")
|
||||||
|
.flatten()
|
||||||
|
}
|
||||||
363
backend/tests/export_epoch.rs
Normal file
363
backend/tests/export_epoch.rs
Normal file
@@ -0,0 +1,363 @@
|
|||||||
|
//! DB-backed integration tests for the export epoch state machine (migration 014).
|
||||||
|
//!
|
||||||
|
//! These run against a REAL Postgres: `#[sqlx::test]` creates a throwaway database per test and
|
||||||
|
//! runs `backend/migrations/` into it, so the schema, the enums, the `UNIQUE (event_id, type)`
|
||||||
|
//! constraint and the `export_current` view are the production ones — not a mock.
|
||||||
|
//!
|
||||||
|
//! THE INVARIANT, from migration 014:
|
||||||
|
//!
|
||||||
|
//! An export is downloadable IFF
|
||||||
|
//! event.export_released_at IS NOT NULL
|
||||||
|
//! AND export_job.epoch = event.export_epoch
|
||||||
|
//! AND export_job.status = 'done'
|
||||||
|
//!
|
||||||
|
//! Readiness is DERIVED (the `export_current` view), never stored. Every test below pins one leg of
|
||||||
|
//! that invariant with the exact SQL `src/` executes (see `tests/common/mod.rs`).
|
||||||
|
|
||||||
|
mod common;
|
||||||
|
|
||||||
|
use common::*;
|
||||||
|
use sqlx::PgPool;
|
||||||
|
|
||||||
|
// ─────────────────────────────────────────────────────────────────────────────
|
||||||
|
// 1. Epoch monotonicity
|
||||||
|
// ─────────────────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
/// Release, reopen and re-release each bump `export_epoch`, and `RETURNING export_epoch` hands the
|
||||||
|
/// caller the POST-increment value.
|
||||||
|
///
|
||||||
|
/// PREVENTS: a worker born with the PRE-increment epoch. It would be inert from the instant it
|
||||||
|
/// started — every one of its writes is `epoch`-guarded, so `claim_job`/`finalize_job` would match
|
||||||
|
/// nothing, the job row would sit at `pending` 0% forever with no live worker, and the host's
|
||||||
|
/// download button would spin and then 404. The keepsake would never be built at all.
|
||||||
|
#[sqlx::test]
|
||||||
|
async fn release_returns_post_increment_epoch(pool: PgPool) {
|
||||||
|
let event_id = seed_event(&pool, "wedding").await;
|
||||||
|
assert_eq!(event_epoch(&pool, event_id).await, 0, "a fresh event starts at epoch 0");
|
||||||
|
|
||||||
|
let released = release_gallery(&pool, "wedding").await.expect("release claims the event");
|
||||||
|
assert_eq!(released, 1, "RETURNING must give the epoch AFTER the +1, not before");
|
||||||
|
assert_eq!(event_epoch(&pool, event_id).await, released, "worker's epoch == event's epoch");
|
||||||
|
|
||||||
|
// The jobs armed by the release carry exactly that epoch — this is what makes the worker's
|
||||||
|
// guarded writes match.
|
||||||
|
for t in ["zip", "html"] {
|
||||||
|
let (status, epoch, _) = job_row(&pool, event_id, t).await.expect("job armed");
|
||||||
|
assert_eq!(status, "pending");
|
||||||
|
assert_eq!(epoch, released, "{t} job must be armed at the epoch the worker was born with");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Epoch is strictly monotonic across the whole release/reopen/re-release cycle, and a second
|
||||||
|
/// release attempt while already released is rejected WITHOUT bumping.
|
||||||
|
///
|
||||||
|
/// PREVENTS: epoch reuse. If a reopen could return the event to an epoch some old `done` row still
|
||||||
|
/// carries, a retired keepsake — one that a guest asked to be taken down from — would silently
|
||||||
|
/// become downloadable again.
|
||||||
|
#[sqlx::test]
|
||||||
|
async fn epoch_is_strictly_monotonic_across_reopen(pool: PgPool) {
|
||||||
|
let event_id = seed_event(&pool, "wedding").await;
|
||||||
|
|
||||||
|
assert_eq!(release_gallery(&pool, "wedding").await, Some(1));
|
||||||
|
|
||||||
|
// A duplicate release is a no-op (`WHERE export_released_at IS NULL`) and must NOT bump.
|
||||||
|
assert_eq!(release_gallery(&pool, "wedding").await, None, "already released");
|
||||||
|
assert_eq!(event_epoch(&pool, event_id).await, 1, "a rejected release must not move the epoch");
|
||||||
|
|
||||||
|
// Reopen retires the generation with ONE write.
|
||||||
|
assert_eq!(open_event(&pool, "wedding").await, 1);
|
||||||
|
assert_eq!(event_epoch(&pool, event_id).await, 2, "reopen bumps");
|
||||||
|
|
||||||
|
// And re-releasing bumps again — never back to 1.
|
||||||
|
assert_eq!(release_gallery(&pool, "wedding").await, Some(3), "re-release bumps again");
|
||||||
|
assert_eq!(event_epoch(&pool, event_id).await, 3);
|
||||||
|
}
|
||||||
|
|
||||||
|
// ─────────────────────────────────────────────────────────────────────────────
|
||||||
|
// 2. A retired-epoch worker is inert
|
||||||
|
// ─────────────────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
/// A worker holding a retired epoch cannot write anything anybody can see: once the rows have been
|
||||||
|
/// re-armed at a newer epoch, its `update_progress` and `finalize_job` both match 0 rows, and
|
||||||
|
/// `export_current` never exposes its output.
|
||||||
|
///
|
||||||
|
/// PREVENTS: the classic lost race — a slow worker from BEFORE a takedown finishing afterwards and
|
||||||
|
/// publishing an archive that still contains the photo a guest asked to have removed. "Please take
|
||||||
|
/// my photo out" is the one request that most needs to reach the keepsake, and the keepsake is the
|
||||||
|
/// artifact people keep forever.
|
||||||
|
#[sqlx::test]
|
||||||
|
async fn retired_epoch_worker_writes_are_no_ops(pool: PgPool) {
|
||||||
|
let event_id = seed_event(&pool, "wedding").await;
|
||||||
|
let old_epoch = release_gallery(&pool, "wedding").await.unwrap();
|
||||||
|
|
||||||
|
// Worker A is born at epoch 1 and claims the ZIP.
|
||||||
|
assert!(claim_job(&pool, event_id, "zip", old_epoch).await, "worker A wins its claim");
|
||||||
|
assert!(update_progress(&pool, event_id, "zip", old_epoch, 40).await, "still live at 40%");
|
||||||
|
|
||||||
|
// ── A takedown lands mid-export: `invalidate_and_arm(Affects::Both)` bumps and re-arms. ──
|
||||||
|
let (_, _, new_epoch) = bump_epoch(&pool, "wedding").await.expect("bump on a released event");
|
||||||
|
assert_eq!(new_epoch, old_epoch + 1);
|
||||||
|
let mut conn = pool.acquire().await.unwrap();
|
||||||
|
enqueue_types_at_epoch(&mut conn, event_id, new_epoch, &["zip", "html"]).await;
|
||||||
|
drop(conn);
|
||||||
|
|
||||||
|
// Worker A is now INERT BY CONSTRUCTION. Every write is guarded on its own birth epoch.
|
||||||
|
assert!(
|
||||||
|
!update_progress(&pool, event_id, "zip", old_epoch, 90).await,
|
||||||
|
"the liveness check must report `false` so worker A stops grinding through the gallery"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
!finalize_job(&pool, event_id, "zip", old_epoch, "exports/Gallery.1.zip").await,
|
||||||
|
"worker A's finalize MUST affect 0 rows — this is the write that would have published a \
|
||||||
|
keepsake still containing the taken-down photo"
|
||||||
|
);
|
||||||
|
|
||||||
|
// The re-armed row is untouched by the loser: still pending at the LIVE epoch, waiting for the
|
||||||
|
// fresh worker. (If worker A had won, this row would read `done` at epoch 1.)
|
||||||
|
let (status, epoch, file_path) = job_row(&pool, event_id, "zip").await.unwrap();
|
||||||
|
assert_eq!((status.as_str(), epoch), ("pending", new_epoch));
|
||||||
|
assert_eq!(file_path, None, "the loser's file_path must never be recorded");
|
||||||
|
|
||||||
|
// And nothing is downloadable — not the stale archive, not anything.
|
||||||
|
assert_eq!(downloadable(&pool, event_id, "zip").await, None);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The documented, deliberate nuance in `claim_job`: after a bare `open_event` (which writes
|
||||||
|
/// NOTHING to `export_job` — that is the point of the design), a worker at the old epoch still WINS
|
||||||
|
/// its claim and can still write `done`. That is wasted work, not incorrectness: retirement is
|
||||||
|
/// enforced at READ time. `export_current` must refuse to expose the row.
|
||||||
|
///
|
||||||
|
/// PREVENTS: someone "optimising" `claim_job` into a cross-table `EXISTS (SELECT ... FROM event)`
|
||||||
|
/// guard — the exact unsound guard migration 014 removed (under READ COMMITTED, a blocked UPDATE
|
||||||
|
/// re-evaluates same-row predicates but answers other-table subqueries from a stale snapshot).
|
||||||
|
/// This test pins the read-time enforcement so the write-time guard is never re-added.
|
||||||
|
#[sqlx::test]
|
||||||
|
async fn reopen_retires_at_read_time_not_write_time(pool: PgPool) {
|
||||||
|
let event_id = seed_event(&pool, "wedding").await;
|
||||||
|
let epoch = release_gallery(&pool, "wedding").await.unwrap();
|
||||||
|
|
||||||
|
assert!(claim_job(&pool, event_id, "zip", epoch).await);
|
||||||
|
|
||||||
|
// Host reopens uploads. No export_job row is touched.
|
||||||
|
assert_eq!(open_event(&pool, "wedding").await, 1);
|
||||||
|
|
||||||
|
// The in-flight worker's row-local writes still match — it was never told to stop.
|
||||||
|
assert!(
|
||||||
|
finalize_job(&pool, event_id, "zip", epoch, "exports/Gallery.1.zip").await,
|
||||||
|
"documented: the claim/finalize is guarded on the JOB row's epoch, not the event's"
|
||||||
|
);
|
||||||
|
let (status, _, _) = job_row(&pool, event_id, "zip").await.unwrap();
|
||||||
|
assert_eq!(status, "done", "the row really does say done");
|
||||||
|
|
||||||
|
// …and yet it is invisible. `export_current` requires the event to be released AND the epochs to
|
||||||
|
// match; the reopen broke both. A worker at a dead epoch writes a row nobody can see.
|
||||||
|
assert!(!in_export_current(&pool, event_id, "zip").await);
|
||||||
|
assert_eq!(
|
||||||
|
downloadable(&pool, event_id, "zip").await,
|
||||||
|
None,
|
||||||
|
"a reopened event must serve NO keepsake, however finished the job row looks"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
// ─────────────────────────────────────────────────────────────────────────────
|
||||||
|
// 3. `export_current` exactness (table-driven)
|
||||||
|
// ─────────────────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
/// The view is the ONE definition of "downloadable". Released + `done` + matching epoch ⇒ present;
|
||||||
|
/// break ANY single leg ⇒ absent. Nothing else may make it appear or disappear.
|
||||||
|
///
|
||||||
|
/// PREVENTS, leg by leg:
|
||||||
|
/// * `released` — serving a keepsake for an event whose uploads are still open, i.e. an archive
|
||||||
|
/// missing every photo taken after the snapshot.
|
||||||
|
/// * `done` — handing out a half-written ZIP (a corrupt keepsake, downloaded once, kept forever).
|
||||||
|
/// * `epoch` — the retired-generation download: the 404-forever keepsake, or worse, the archive
|
||||||
|
/// still containing content that was taken down.
|
||||||
|
#[sqlx::test]
|
||||||
|
async fn export_current_is_exactly_the_invariant(pool: PgPool) {
|
||||||
|
// (name, released?, status, job epoch offset from the event epoch, expected visible)
|
||||||
|
let cases: &[(&str, bool, &str, i64, bool)] = &[
|
||||||
|
("released + done + current epoch", true, "done", 0, true),
|
||||||
|
("NOT released (done, epoch matches)", false, "done", 0, false),
|
||||||
|
("NOT done: pending", true, "pending", 0, false),
|
||||||
|
("NOT done: running", true, "running", 0, false),
|
||||||
|
("NOT done: failed", true, "failed", 0, false),
|
||||||
|
("stale epoch (done, released)", true, "done", -1, false),
|
||||||
|
("future epoch (done, released)", true, "done", 1, false),
|
||||||
|
("migration-014 retired sentinel epoch -1", true, "done", -2, false),
|
||||||
|
];
|
||||||
|
|
||||||
|
for (i, (name, released, status, offset, expect_visible)) in cases.iter().enumerate() {
|
||||||
|
let slug = format!("case{i}");
|
||||||
|
let event_id = seed_event(&pool, &slug).await;
|
||||||
|
|
||||||
|
// Get the event to a known epoch (1) either by releasing it, or — for the unreleased case —
|
||||||
|
// by releasing and reopening, which leaves it unreleased at a non-zero epoch.
|
||||||
|
let event_epoch_now = if *released {
|
||||||
|
release_gallery(&pool, &slug).await.unwrap()
|
||||||
|
} else {
|
||||||
|
release_gallery(&pool, &slug).await.unwrap();
|
||||||
|
open_event(&pool, &slug).await;
|
||||||
|
event_epoch(&pool, event_id).await
|
||||||
|
};
|
||||||
|
|
||||||
|
// Plant a single ZIP job row in the exact state under test. `-2` encodes "the sentinel the
|
||||||
|
// migration stamps on retired rows", which must never equal a non-negative event epoch.
|
||||||
|
// (Clear the rows the release armed first — `UNIQUE (event_id, type)`.)
|
||||||
|
sqlx::query("DELETE FROM export_job WHERE event_id = $1")
|
||||||
|
.bind(event_id)
|
||||||
|
.execute(&pool)
|
||||||
|
.await
|
||||||
|
.expect("clear armed jobs");
|
||||||
|
|
||||||
|
let job_epoch = if *offset == -2 { -1 } else { event_epoch_now + offset };
|
||||||
|
sqlx::query(
|
||||||
|
"INSERT INTO export_job (event_id, type, status, progress_pct, epoch, file_path)
|
||||||
|
VALUES ($1, 'zip', $2::export_status, 100, $3, 'exports/Gallery.zip')",
|
||||||
|
)
|
||||||
|
.bind(event_id)
|
||||||
|
.bind(*status)
|
||||||
|
.bind(job_epoch)
|
||||||
|
.execute(&pool)
|
||||||
|
.await
|
||||||
|
.expect("plant job row");
|
||||||
|
|
||||||
|
let visible = downloadable(&pool, event_id, "zip").await.is_some();
|
||||||
|
assert_eq!(
|
||||||
|
visible, *expect_visible,
|
||||||
|
"export_current exactness violated for case: {name} \
|
||||||
|
(released={released}, status={status}, job_epoch={job_epoch}, event_epoch={event_epoch_now})"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ─────────────────────────────────────────────────────────────────────────────
|
||||||
|
// 4. The ViewerOnly ZIP carry-forward
|
||||||
|
// ─────────────────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
/// Branch A — the ZIP is `done` at the outgoing epoch: the carry-forward re-stamps it to the new
|
||||||
|
/// epoch (rows_affected = 1), so only the HTML viewer is rebuilt and the finished ZIP stays
|
||||||
|
/// downloadable throughout.
|
||||||
|
///
|
||||||
|
/// PREVENTS: rebuilding a multi-GB archive because someone deleted a comment. The ZIP holds media,
|
||||||
|
/// not comments — a needless rebuild would 404 the photo download for minutes to change nothing
|
||||||
|
/// inside it.
|
||||||
|
#[sqlx::test]
|
||||||
|
async fn viewer_only_carries_a_done_zip_forward(pool: PgPool) {
|
||||||
|
let event_id = seed_event(&pool, "wedding").await;
|
||||||
|
let e1 = release_gallery(&pool, "wedding").await.unwrap();
|
||||||
|
|
||||||
|
// Both halves finish at epoch 1 — the keepsake is live.
|
||||||
|
for t in ["zip", "html"] {
|
||||||
|
assert!(claim_job(&pool, event_id, t, e1).await);
|
||||||
|
assert!(finalize_job(&pool, event_id, t, e1, &format!("exports/{t}.{e1}.zip")).await);
|
||||||
|
}
|
||||||
|
let zip_file = downloadable(&pool, event_id, "zip").await.expect("zip is live");
|
||||||
|
|
||||||
|
// ── A comment is moderated: invalidate_and_arm(Affects::ViewerOnly). ──
|
||||||
|
let (_, _, e2) = bump_epoch(&pool, "wedding").await.unwrap();
|
||||||
|
let carried = carry_zip_forward(&pool, event_id, e2).await;
|
||||||
|
assert!(carried, "a `done` ZIP at epoch-1 MUST be carried forward (rows_affected == 1)");
|
||||||
|
|
||||||
|
// Only the viewer is re-armed…
|
||||||
|
let mut conn = pool.acquire().await.unwrap();
|
||||||
|
enqueue_types_at_epoch(&mut conn, event_id, e2, &["html"]).await;
|
||||||
|
drop(conn);
|
||||||
|
|
||||||
|
// …and the ZIP is STILL DOWNLOADABLE, at the new epoch, pointing at the same, unrenamed file.
|
||||||
|
let (status, epoch, _) = job_row(&pool, event_id, "zip").await.unwrap();
|
||||||
|
assert_eq!((status.as_str(), epoch), ("done", e2), "the ZIP row rode the epoch bump");
|
||||||
|
assert_eq!(
|
||||||
|
downloadable(&pool, event_id, "zip").await,
|
||||||
|
Some(zip_file),
|
||||||
|
"the carried archive must never stop being served — same file, new epoch"
|
||||||
|
);
|
||||||
|
|
||||||
|
// The viewer, meanwhile, is correctly retired and pending a rebuild.
|
||||||
|
assert_eq!(job_row(&pool, event_id, "html").await.unwrap().0, "pending");
|
||||||
|
assert_eq!(downloadable(&pool, event_id, "html").await, None);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Branch B — THE BUG WE JUST FIXED. If the ZIP is still `pending`/`running` when the comment is
|
||||||
|
/// moderated (which is MINUTES for a real multi-GB gallery, and deleting a comment right after
|
||||||
|
/// release is an utterly ordinary thing to do), the carry-forward matches NOTHING
|
||||||
|
/// (rows_affected = 0) — so the caller must NOT assume it carried, and must re-arm the ZIP too.
|
||||||
|
///
|
||||||
|
/// PREVENTS: the stranded ZIP. Blindly re-arming only the viewer would leave the ZIP row at the
|
||||||
|
/// retired epoch; the in-flight worker then finishes and writes `done` at an epoch `export_current`
|
||||||
|
/// no longer matches, nothing ever re-arms it, and `GET /export/zip` 404s FOREVER — a keepsake the
|
||||||
|
/// couple paid for that simply never appears, short of a reboot.
|
||||||
|
#[sqlx::test]
|
||||||
|
async fn viewer_only_carry_forward_matches_nothing_when_zip_unfinished(pool: PgPool) {
|
||||||
|
for zip_state in ["pending", "running"] {
|
||||||
|
let slug = format!("wedding-{zip_state}");
|
||||||
|
let event_id = seed_event(&pool, &slug).await;
|
||||||
|
let e1 = release_gallery(&pool, &slug).await.unwrap();
|
||||||
|
|
||||||
|
// The ZIP worker is still going; only the viewer has finished.
|
||||||
|
if zip_state == "running" {
|
||||||
|
assert!(claim_job(&pool, event_id, "zip", e1).await);
|
||||||
|
}
|
||||||
|
assert!(claim_job(&pool, event_id, "html", e1).await);
|
||||||
|
assert!(finalize_job(&pool, event_id, "html", e1, "exports/Memories.1.zip").await);
|
||||||
|
|
||||||
|
// ── The comment is moderated. ──
|
||||||
|
let (_, _, e2) = bump_epoch(&pool, &slug).await.unwrap();
|
||||||
|
let carried = carry_zip_forward(&pool, event_id, e2).await;
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
!carried,
|
||||||
|
"a {zip_state} ZIP has nothing to carry forward — the UPDATE must affect 0 rows \
|
||||||
|
(its `status = 'done'` predicate is the whole precondition)"
|
||||||
|
);
|
||||||
|
|
||||||
|
// The carry-forward's OWN result decides. It didn't match ⇒ rebuild the ZIP as well.
|
||||||
|
let types: &[&str] = if carried { &["html"] } else { &["zip", "html"] };
|
||||||
|
let mut conn = pool.acquire().await.unwrap();
|
||||||
|
enqueue_types_at_epoch(&mut conn, event_id, e2, types).await;
|
||||||
|
drop(conn);
|
||||||
|
|
||||||
|
// THE ASSERTION THAT WOULD HAVE CAUGHT THE BUG: the ZIP must not be stranded at the dead
|
||||||
|
// epoch. It is re-armed at the live one, so a fresh worker will actually build it.
|
||||||
|
let (status, epoch, _) = job_row(&pool, event_id, "zip").await.unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
(status.as_str(), epoch),
|
||||||
|
("pending", e2),
|
||||||
|
"the unfinished ZIP MUST be re-armed at the new epoch, not left stranded at {e1}"
|
||||||
|
);
|
||||||
|
|
||||||
|
// Even if the old in-flight worker now "finishes", it is inert and cannot resurrect itself.
|
||||||
|
assert!(!finalize_job(&pool, event_id, "zip", e1, "exports/Gallery.1.zip").await);
|
||||||
|
assert_eq!(downloadable(&pool, event_id, "zip").await, None);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The re-arm upsert must never clobber the archive it just carried forward.
|
||||||
|
///
|
||||||
|
/// `enqueue_types_at_epoch`'s `WHERE export_job.status <> 'done' OR export_job.epoch <> EXCLUDED.epoch`
|
||||||
|
/// is the "startup recovery must not clobber a good half" rule, expressed as the readiness predicate
|
||||||
|
/// itself. PREVENTS: boot recovery resetting a perfectly good, downloadable ZIP back to `pending`
|
||||||
|
/// and making the keepsake 404 while it needlessly rebuilds.
|
||||||
|
#[sqlx::test]
|
||||||
|
async fn enqueue_preserves_a_done_half_at_the_current_epoch(pool: PgPool) {
|
||||||
|
let event_id = seed_event(&pool, "wedding").await;
|
||||||
|
let e1 = release_gallery(&pool, "wedding").await.unwrap();
|
||||||
|
|
||||||
|
// The ZIP finished; the HTML worker was killed mid-flight (crash) and sits at `running`.
|
||||||
|
assert!(claim_job(&pool, event_id, "zip", e1).await);
|
||||||
|
assert!(finalize_job(&pool, event_id, "zip", e1, "exports/Gallery.1.zip").await);
|
||||||
|
assert!(claim_job(&pool, event_id, "html", e1).await);
|
||||||
|
|
||||||
|
// Boot recovery re-arms both types at the SAME epoch.
|
||||||
|
let mut conn = pool.acquire().await.unwrap();
|
||||||
|
enqueue_types_at_epoch(&mut conn, event_id, e1, &["zip", "html"]).await;
|
||||||
|
drop(conn);
|
||||||
|
|
||||||
|
// The good half survives untouched…
|
||||||
|
let (status, epoch, file_path) = job_row(&pool, event_id, "zip").await.unwrap();
|
||||||
|
assert_eq!((status.as_str(), epoch), ("done", e1), "a done half at the live epoch is preserved");
|
||||||
|
assert_eq!(file_path.as_deref(), Some("exports/Gallery.1.zip"), "file_path not nulled");
|
||||||
|
assert!(downloadable(&pool, event_id, "zip").await.is_some());
|
||||||
|
|
||||||
|
// …and only the missing half is re-armed.
|
||||||
|
assert_eq!(job_row(&pool, event_id, "html").await.unwrap().0, "pending");
|
||||||
|
}
|
||||||
268
backend/tests/upload_concurrency.rs
Normal file
268
backend/tests/upload_concurrency.rs
Normal file
@@ -0,0 +1,268 @@
|
|||||||
|
//! 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();
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user