diff --git a/backend/src/handlers/upload.rs b/backend/src/handlers/upload.rs index d0c8573..b30b964 100644 --- a/backend/src/handlers/upload.rs +++ b/backend/src/handlers/upload.rs @@ -172,6 +172,74 @@ impl Drop for TempFileGuard { } } +/// Guarantees the compression job is queued even if this request never finishes. +/// +/// The enqueue is a bare `tokio::spawn` that happens AFTER `tx.commit()`, and the commit is a +/// suspension point. If the guest walks out of range inside it, Postgres applies the COMMIT and +/// axum drops this future before the spawn is reached: the row lands durably at +/// `compression_status = 'pending'` with no job behind it, and nothing in the app ever looks at +/// it again — `startup_recovery` only rescues `'processing'`, and the derivative backfill runs +/// once at boot. The photo then shows in the feed with no preview, forces every viewer to pull +/// the full original, and is skipped by the diashow forever. +/// +/// This is the same hazard `TempFileGuard` covers for the file, reached one line later, and it +/// gets the same answer: `Drop` runs on cancellation, so arming before the commit and disarming +/// after the spawn closes the window. It also covers the indeterminate-commit error path, which +/// returns `Err` on a row that may well be live. +/// +/// Enqueuing a job for a row that did NOT commit is harmless: `set_compression_status` matches +/// zero rows and `begin_derivative_attempt` returns `None`, so the task retires immediately. +/// A guard is still not sufficient on its own — nothing in-process survives a SIGKILL — which is +/// why `CompressionWorker::requeue_stuck_pending` sweeps for the same state on a timer. +struct EnqueueGuard { + /// `None` once the job has actually been handed to the worker. + armed: Option<(crate::services::compression::CompressionWorker, Uuid, String, String)>, +} + +impl EnqueueGuard { + fn new( + worker: crate::services::compression::CompressionWorker, + upload_id: Uuid, + original_path: String, + mime_type: String, + ) -> Self { + Self { + armed: Some((worker, upload_id, original_path, mime_type)), + } + } + + /// The job is queued; stand down. + fn disarm(&mut self) { + self.armed = None; + } +} + +impl Drop for EnqueueGuard { + fn drop(&mut self) { + let Some((worker, upload_id, path, mime)) = self.armed.take() else { + return; + }; + // `process` calls `tokio::spawn`, which panics without a runtime. During a normal + // cancellation we are still on the runtime that dropped us; during shutdown we may not + // be. Log loudly rather than panicking in a destructor — the timed sweeper is the + // backstop for exactly this case. + if tokio::runtime::Handle::try_current().is_err() { + tracing::error!( + %upload_id, + "upload committed but the compression job could not be queued (no runtime in \ + Drop); requeue_stuck_pending will pick it up" + ); + return; + } + tracing::warn!( + %upload_id, + "request ended before the compression job was queued — queueing it from the drop \ + guard (client most likely disconnected during COMMIT)" + ); + worker.process(upload_id, path, mime); + } +} + /// Allowlist of accepted media types, keyed by the MIME that `infer` derives from /// the file's magic bytes. The detected MIME (not the client-declared one) is what /// we trust, store, and hand to the compression pipeline — so a text-based payload @@ -719,7 +787,9 @@ pub async fn upload( // Quota accounting, the upload row, and its hashtag links must be atomic: a // crash between the bytes increment and the insert would permanently charge // bytes with no row to reclaim them (silent quota erosion / spurious lockout). - let tx_result: Result = async { + // Carries the enqueue guard out with the row: it is armed inside the block (before the + // commit) and can only be stood down after `process()` runs, which happens out here. + let tx_result: Result<(Upload, EnqueueGuard), AppError> = async { let mut tx = state.pool.begin().await?; // RE-CHECK THE LOCK, UNDER A ROW LOCK, INSIDE THE COMMIT TX. @@ -836,6 +906,15 @@ pub async fn upload( // the failure to the recoverable side: if we are dropped mid-commit the bytes leak, // and leaked bytes under a final name are exactly what the orphan sweeper reclaims. // A committed row whose file we deleted is unrecoverable. Prefer the leak. + // Armed BEFORE the commit, for the mirror-image of the reason the file guard is + // disarmed before it: from here on, a cancellation can leave a live row behind, and a + // live row with no compression job is silent, permanent loss. See `EnqueueGuard`. + let enqueue_guard = EnqueueGuard::new( + state.compression.clone(), + upload.id, + relative_path.clone(), + mime.clone(), + ); file_guard.disarm(); if let Err(e) = tx.commit().await { // Deliberately do NOT re-arm the guard here. @@ -861,7 +940,7 @@ pub async fn upload( ); return Err(e.into()); } - Ok(upload) + Ok((upload, enqueue_guard)) } .await; @@ -873,8 +952,8 @@ pub async fn upload( // // The successful-commit case disarmed the guard inside the block, immediately before // `tx.commit()` — see the comment there for why it cannot be done out here. - let upload = match tx_result { - Ok(u) => u, + let (upload, mut enqueue_guard) = match tx_result { + Ok(v) => v, // 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. The // loser's bytes are reclaimed by the guard when this return drops it. @@ -917,10 +996,12 @@ pub async fn upload( Err(e) => return Err(e), }; - // Spawn compression task + // Spawn compression task. The guard above has covered this call since before the commit; + // stand it down only once the job is genuinely with the worker. state .compression .process(upload.id, relative_path, mime.clone()); + enqueue_guard.disarm(); // Broadcast SSE event let dto = UploadDto { diff --git a/backend/src/main.rs b/backend/src/main.rs index 4f28926..365414e 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -127,6 +127,7 @@ async fn main() -> Result<()> { state.rate_limiter.clone(), state.sse_tickets.clone(), config.media_path.clone(), + state.compression.clone(), ); let api = Router::new() diff --git a/backend/src/services/compression.rs b/backend/src/services/compression.rs index 285df96..55a3db6 100644 --- a/backend/src/services/compression.rs +++ b/backend/src/services/compression.rs @@ -411,6 +411,66 @@ impl CompressionWorker { /// `derivative_attempts` stops a fatal row being replayed on every boot, `BACKFILL_BATCH` /// stops one start queueing unbounded work, and the whole thing runs as ONE task walking /// the rows sequentially rather than N tasks racing for the same semaphore. + /// How long an upload may sit at `'pending'` before the sweeper assumes its job was lost. + /// + /// Must comfortably exceed the deepest real queue wait, or the sweeper re-enqueues photos + /// that are merely waiting their turn. The live path makes that easy to bound: a task's + /// FIRST action after taking a permit is to flip the row to `'processing'`, so a row still + /// `'pending'` after this long is either behind that much work or genuinely lost. The + /// event simulation peaked at 120 queued with a p99 of 97s while being fed thirty times a + /// real event's arrival rate, so ten minutes is a wide margin rather than a tight one. + const PENDING_GRACE: chrono::Duration = chrono::Duration::minutes(10); + + /// Re-queue uploads whose compression job was lost between the commit and the spawn. + /// + /// `EnqueueGuard` closes that window for a cancelled request, but nothing in-process + /// survives a SIGKILL or an OOM kill, and `startup_recovery` only rescues `'processing'` + /// — so a row orphaned at `'pending'` had no path back at all. Without this, the failure is + /// silent and permanent: the photo keeps its feed entry, never gets a derivative, and is + /// dropped from the diashow for the rest of the event. + /// + /// Re-entering the live `process` path (rather than regenerating derivatives inline, as + /// `backfill_stale_derivatives` does) is deliberate: it reuses the status transitions, the + /// retry ladder and the SSE broadcast, so a rescued photo behaves exactly like one that was + /// never lost. + pub async fn requeue_stuck_pending(&self) { + let cutoff = chrono::Utc::now() - Self::PENDING_GRACE; + let rows = sqlx::query_as::<_, (Uuid, String, String)>( + "SELECT id, original_path, mime_type FROM upload + WHERE deleted_at IS NULL + AND compression_status = 'pending' + AND original_path <> '' + AND created_at < $1 + AND derivative_attempts < $2 + ORDER BY created_at + LIMIT $3", + ) + .bind(cutoff) + .bind(Self::MAX_DERIVATIVE_ATTEMPTS) + .bind(Self::BACKFILL_BATCH) + .fetch_all(&self.pool) + .await; + let rows = match rows { + Ok(r) => r, + Err(e) => { + tracing::warn!(error = ?e, "stuck-pending sweep query failed"); + return; + } + }; + if rows.is_empty() { + return; + } + // WARN, not INFO: reaching this means the guard did not run, which is worth noticing + // even though the photo is being rescued. + tracing::warn!( + count = rows.len(), + "found upload(s) stuck at 'pending' with no compression job; re-queueing" + ); + for (id, original_path, mime_type) in rows { + self.process(id, original_path, mime_type); + } + } + pub async fn backfill_stale_derivatives(&self) { // `original_path IS NOT NULL` was dead — the column is NOT NULL. What actually needs // excluding is the blanked path `cleanup_deleted_media` leaves behind. diff --git a/backend/src/services/maintenance.rs b/backend/src/services/maintenance.rs index e694ab8..451d82e 100644 --- a/backend/src/services/maintenance.rs +++ b/backend/src/services/maintenance.rs @@ -144,7 +144,36 @@ pub fn spawn_periodic_tasks( rate_limiter: RateLimiter, sse_tickets: SseTicketStore, media_path: PathBuf, + compression: crate::services::compression::CompressionWorker, ) { + // Rescuing a lost compression job gets its OWN loop rather than a line in the hourly one. + // Cadence is the whole point: an upload stuck at 'pending' is a photo missing from the + // diashow and showing without a preview, so an hour of that during a five-hour party is a + // guest-visible hole. Ten minutes bounds it while still sitting far above the deepest + // observed queue wait (see `PENDING_GRACE`). Supervised for the same reason as the loop + // below: silently dying is how this class of safety net stops existing. + tokio::spawn(async move { + loop { + let worker = compression.clone(); + let inner = tokio::spawn(async move { + // `interval` fires its first tick immediately, and that is deliberate here: + // a restart is exactly when stranded rows exist (the drop guard cannot + // survive a SIGKILL), and `startup_recovery` rescues only 'processing'. The + // grace window makes the boot pass safe — nothing uploaded in the last ten + // minutes can match, so this can never race a job that is merely queued. + let mut tick = tokio::time::interval(Duration::from_secs(600)); + loop { + tick.tick().await; + worker.requeue_stuck_pending().await; + } + }); + match inner.await { + Ok(()) => tracing::error!("stuck-pending sweeper returned; restarting it"), + Err(e) => tracing::error!(error = ?e, "stuck-pending sweeper died; restarting it"), + } + tokio::time::sleep(Duration::from_secs(60)).await; + } + }); // Supervised, because this one task carries EVERY piece of recurring hygiene in the app: // session pruning, media reclamation, the orphan-temp sweep, and the rate-limiter and // SSE-ticket maps. As a bare `tokio::spawn` with no retained handle, a single panic anywhere