diff --git a/backend/src/handlers/test_admin.rs b/backend/src/handlers/test_admin.rs index b6c40ce..917ccbf 100644 --- a/backend/src/handlers/test_admin.rs +++ b/backend/src/handlers/test_admin.rs @@ -98,6 +98,13 @@ pub async fn truncate_all( // surviving ticket is a dangling reference to a user that no longer exists. state.sse_tickets.clear(); + // Invalidate any in-flight/queued compression task spawned by the previous test. Without this a + // task still waiting on the concurrency semaphore wakes AFTER this wipe, fails to find its + // (now-deleted) file, and broadcasts upload-error/upload-deleted into the NEXT test's SSE + // stream. (Export workers are already inert across a truncate: they are epoch-guarded on the + // event row, and truncate gives the event a fresh random UUID, so their writes match nothing.) + state.compression.bump_generation(); + Ok(StatusCode::NO_CONTENT) } diff --git a/backend/src/handlers/upload.rs b/backend/src/handlers/upload.rs index 0edcc5a..e0fdbd2 100644 --- a/backend/src/handlers/upload.rs +++ b/backend/src/handlers/upload.rs @@ -416,6 +416,15 @@ pub async fn edit_upload( // Caption update + hashtag wipe-then-relink in one transaction, so a crash // mid-relink can't leave the upload with its hashtags stripped. + // + // Editing is intentionally allowed while uploads are locked or the gallery is released — like + // comments and likes, the lock freezes *new uploads* only (USER_JOURNEYS §9.3). But a caption + // is embedded in the HTML viewer keepsake (the ZIP holds media only — see export.rs), so an + // edit AFTER release must regenerate the viewer, or the downloadable keepsake keeps showing the + // old caption forever while the live feed shows the new one. Same atomicity as delete_upload: + // the edit and its invalidation share one tx so a dropped handler can't leave them disagreeing. + // `Affects::ViewerOnly` carries the finished ZIP forward (the media didn't change); when the + // gallery isn't released, `invalidate_and_arm` returns None and this is a no-op. let mut tx = state.pool.begin().await?; if let Some(ref caption) = body.caption { Upload::update_caption(&mut *tx, upload_id, Some(caption)).await?; @@ -427,7 +436,16 @@ pub async fn edit_upload( Hashtag::link_to_upload(&mut *tx, upload_id, h.id).await?; } } + let regen = crate::services::export::invalidate_and_arm( + &mut tx, + &state.config.event_slug, + crate::services::export::Affects::ViewerOnly, + ) + .await?; tx.commit().await?; + if let Some(r) = regen { + crate::handlers::host::start_regen(&state, r); + } Ok(StatusCode::OK) } diff --git a/backend/src/services/compression.rs b/backend/src/services/compression.rs index 44813c0..29fcb1b 100644 --- a/backend/src/services/compression.rs +++ b/backend/src/services/compression.rs @@ -1,4 +1,5 @@ use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use anyhow::{Context, Result}; @@ -15,6 +16,10 @@ pub struct CompressionWorker { pool: PgPool, media_path: PathBuf, sse_tx: broadcast::Sender, + /// Bumped whenever the underlying data is reset out from under in-flight work (only the e2e + /// TRUNCATE does this today). A task captures the value at spawn and abandons itself if it has + /// changed by the time it runs — see `process`. + generation: Arc, } impl CompressionWorker { @@ -24,14 +29,32 @@ impl CompressionWorker { pool, media_path, sse_tx, + generation: Arc::new(AtomicU64::new(0)), } } + /// Invalidate all in-flight and queued compression work. Called by the e2e TRUNCATE endpoint: + /// truncating deletes the upload rows and wipes `media/`, so a worker that was queued on the + /// semaphore when the wipe happened would otherwise wake in the NEXT test, fail to find its + /// file, and broadcast `upload-error` / `upload-deleted` into that test's live SSE stream — + /// corrupting any test that asserts on toasts or feed contents. Bumping the generation makes + /// those stale tasks return silently instead. A no-op in production (never called there). + pub fn bump_generation(&self) { + self.generation.fetch_add(1, Ordering::SeqCst); + } + /// Spawn a background task to process an uploaded file. pub fn process(&self, upload_id: Uuid, original_path: String, mime_type: String) { let worker = self.clone(); + let born_at = worker.generation.load(Ordering::SeqCst); tokio::spawn(async move { let _permit = worker.semaphore.acquire().await; + // The data this task was queued against may have been reset while it waited for a permit + // (e2e TRUNCATE). If so, its file and row are gone; doing anything — including + // broadcasting a failure — would leak into an unrelated test. Abandon quietly. + if worker.generation.load(Ordering::SeqCst) != born_at { + return; + } match worker.do_process(upload_id, &original_path, &mime_type).await { Ok(_) => { tracing::info!("compression completed for upload {upload_id}");