Skip to main content

max / makenotwork

41.0 KB · 968 lines History Blame Raw
1 //! Scheduled cleanup jobs: sandbox expiry, terminated accounts, content
2 //! removal, IP scrubbing, stale pending transactions, orphaned uploads, cart
3 //! items, soft-deleted item purges, and pending S3 deletion retries.
4
5 use crate::AppState;
6 use crate::constants;
7 use crate::db;
8 use crate::storage::S3DeleteAuthority;
9
10 /// Hourly: purge `scan_jobs` rows in a terminal state older than the
11 /// retention window. Queued/running rows are not touched.
12 #[tracing::instrument(skip_all, name = "scheduler::purge_old_scan_jobs")]
13 pub(super) async fn purge_old_scan_jobs(state: &AppState) {
14 let window = chrono::Duration::days(constants::SCAN_JOB_RETENTION_DAYS as i64);
15 match db::scan_jobs::purge_old_terminal(&state.db, window).await {
16 Ok(n) => {
17 if n > 0 {
18 tracing::info!(purged = n, "scan_jobs retention sweep");
19 }
20 let _ = db::scheduler_jobs::record_job_run(&state.db, "scan_jobs_retention", n as i64)
21 .await;
22 }
23 Err(e) => tracing::error!(error = ?e, "scan_jobs retention sweep failed"),
24 }
25 }
26
27 /// Drain `ids` through `cleanup_user_s3_and_delete` with bounded concurrency
28 /// (`CLEANUP_PARALLELISM` in flight). Returns the count of successful deletes.
29 ///
30 /// The single bounded-drain primitive shared by every per-user cleanup sweep,
31 /// each cleanup holds a few pool conns + an S3 delete-prefix, so the cap (4)
32 /// leaves the bulk of the ~25-conn pool for request handlers even on a
33 /// mass-expiry tick. One copy means the cap cannot be forgotten or drift
34 /// between sweeps.
35 async fn drain_cleanup_bounded(
36 state: &AppState,
37 ids: Vec<db::UserId>,
38 event: &'static str,
39 label: &'static str,
40 ) -> i64 {
41 const CLEANUP_PARALLELISM: usize = 4;
42 let mut set = tokio::task::JoinSet::new();
43 let mut iter = ids.into_iter();
44 let mut deleted = 0i64;
45 loop {
46 while set.len() < CLEANUP_PARALLELISM {
47 match iter.next() {
48 Some(uid) => {
49 let state = state.clone();
50 set.spawn(async move {
51 cleanup_user_s3_and_delete(&state, uid, event, label).await
52 });
53 }
54 None => break,
55 }
56 }
57 match set.join_next().await {
58 Some(Ok(true)) => deleted += 1,
59 Some(Ok(false)) => {}
60 Some(Err(e)) => tracing::warn!(error = ?e, %label, "cleanup task panicked"),
61 None => break,
62 }
63 }
64 deleted
65 }
66
67 /// Delete expired sandbox accounts and their S3 objects.
68 #[tracing::instrument(skip_all, name = "scheduler::cleanup_sandbox_accounts")]
69 pub(super) async fn cleanup_sandbox_accounts(state: &AppState) {
70 let expired_ids = match db::users::get_expired_sandbox_ids(&state.db).await {
71 Ok(ids) => ids,
72 Err(e) => {
73 tracing::error!(error = ?e, "failed to query expired sandbox accounts");
74 return;
75 }
76 };
77
78 // Drained inline on the tick (sandbox sweep is cheap + frequent); bounded at
79 // 4 by the shared helper.
80 let deleted = drain_cleanup_bounded(state, expired_ids, "sandbox_expired", "sandbox").await;
81 let _ = db::scheduler_jobs::record_job_run(&state.db, "sandbox_cleanup", deleted).await;
82 }
83
84 /// Clean up a user's S3 objects, git repos, and CASCADE-delete the user row.
85 ///
86 /// Shared between sandbox, terminated, and content-removal account cleanup.
87 /// S3 objects are deleted first (before CASCADE removes the DB rows that reference them).
88 async fn cleanup_user_s3_and_delete(
89 state: &AppState,
90 user_id: db::UserId,
91 event: &str,
92 label: &str,
93 ) -> bool {
94 // Resolve everything we need from the DB once, the enqueue list and the
95 // delete list must come from the same snapshot, or a project/app created
96 // between calls will be enqueued but not deleted (or vice versa).
97 // A transient DB error here must ABORT, not degrade to an empty list: an
98 // empty key set would enqueue nothing for `projects/{pid}/` / `{app_id}/`,
99 // then the CASCADE delete below would drop the rows and leave those S3
100 // objects orphaned forever. Return false so the sweep retries next tick
101 // (ultra-fuzz Run 13 Storage: orphan-on-transient-error).
102 let project_ids = match db::projects::get_project_ids_for_user(&state.db, user_id).await {
103 Ok(v) => v,
104 Err(e) => {
105 tracing::error!(error = ?e, %user_id, "{label}: failed to list projects; aborting cleanup to avoid orphaning S3 objects");
106 return false;
107 }
108 };
109 let sync_apps = match db::synckit::get_sync_apps_by_creator(&state.db, user_id).await {
110 Ok(v) => v,
111 Err(e) => {
112 tracing::error!(error = ?e, %user_id, "{label}: failed to list sync apps; aborting cleanup to avoid orphaning S3 objects");
113 return false;
114 }
115 };
116
117 let user_prefix = format!("{user_id}/");
118 let main = crate::storage::S3Bucket::Main.as_str().to_string();
119 let synckit = crate::storage::S3Bucket::Synckit.as_str().to_string();
120 let public = crate::storage::S3Bucket::Public.as_str().to_string();
121 let mut keys: Vec<(String, String)> = Vec::new();
122 keys.push((user_prefix.clone(), main.clone()));
123 // The user's CDN-served image content (`{user_id}/c/{sha}`) lives in the
124 // public bucket, same key prefix, different bucket, so sweep it there too
125 // or those objects orphan on account deletion.
126 keys.push((user_prefix.clone(), public.clone()));
127 for pid in &project_ids {
128 keys.push((format!("projects/{pid}/"), main.clone()));
129 }
130 for app in &sync_apps {
131 keys.push((format!("{}/", app.id), synckit.clone()));
132 keys.push((format!("ota/{}/", app.id), synckit.clone()));
133 }
134
135 // Enqueue all keys before any destructive work
136 if let Err(e) = db::pending_s3_deletions::enqueue_deletions(&state.db, &keys, label).await {
137 tracing::error!(error = ?e, %user_id, "{label}: failed to enqueue S3 deletions, aborting cleanup");
138 return false;
139 }
140
141 let auth = S3DeleteAuthority::new();
142 if let Some(ref s3) = state.storage.s3 {
143 if let Err(e) = s3.delete_prefix(&auth, &user_prefix).await {
144 tracing::warn!(error = ?e, %user_id, "{label}: failed to delete user S3 objects");
145 }
146 for pid in &project_ids {
147 let proj_prefix = format!("projects/{pid}/");
148 if let Err(e) = s3.delete_prefix(&auth, &proj_prefix).await {
149 tracing::warn!(error = ?e, %user_id, %pid, "{label}: failed to delete project S3 objects");
150 }
151 }
152 }
153
154 // The user's CDN-served image content lives under the same `{user_id}/`
155 // prefix in the public bucket; sweep it there too (the enqueue above is the
156 // durable backstop if this fast path fails).
157 if let Some(ref public_s3) = state.storage.public_s3
158 && let Err(e) = public_s3.delete_prefix(&auth, &user_prefix).await
159 {
160 tracing::warn!(error = ?e, %user_id, "{label}: failed to delete user public-bucket objects");
161 }
162
163 if let Some(ref synckit_s3) = state.storage.synckit_s3 {
164 for app in &sync_apps {
165 let blob_prefix = format!("{}/", app.id);
166 if let Err(e) = synckit_s3.delete_prefix(&auth, &blob_prefix).await {
167 tracing::warn!(error = ?e, %user_id, app_id = %app.id, "{label}: failed to delete SyncKit blobs");
168 }
169 let ota_prefix = format!("ota/{}/", app.id);
170 if let Err(e) = synckit_s3.delete_prefix(&auth, &ota_prefix).await {
171 tracing::warn!(error = ?e, %user_id, app_id = %app.id, "{label}: failed to delete OTA artifacts");
172 }
173 }
174 }
175
176 // Git repos on disk
177 if let Some(ref git_root) = state.config.build.git_repos_path
178 && let Ok(Some(user)) = db::users::get_user_by_id(&state.db, user_id).await
179 {
180 cleanup_git_repos_on_disk(git_root, &user.username, user_id).await;
181 }
182
183 // CASCADE delete user row (+ purge domain_cache via the coupled entry point)
184 if let Err(e) = crate::delete_user_account(&state.db, &state.caches, user_id).await {
185 tracing::error!(error = ?e, %user_id, "{label}: failed to delete account");
186 false
187 } else {
188 tracing::info!(%user_id, event, "{label}: account cleaned up");
189 true
190 }
191 }
192
193 /// Remove a user's bare git repositories from disk.
194 ///
195 /// Must be called before `delete_user` (which CASCADE-deletes the git_repos rows).
196 /// Best-effort: logs warnings on failure but does not block account deletion.
197 /// Runs blocking I/O on a dedicated thread to avoid stalling the Tokio runtime.
198 #[tracing::instrument(skip_all, name = "scheduler::cleanup_git_repos_on_disk")]
199 pub(super) async fn cleanup_git_repos_on_disk(
200 git_repos_path: &str,
201 username: &str,
202 user_id: db::UserId,
203 ) {
204 let user_git_dir = std::path::Path::new(git_repos_path).join(username);
205 if user_git_dir.exists() {
206 let path = user_git_dir.clone();
207 match tokio::task::spawn_blocking(move || std::fs::remove_dir_all(&path)).await {
208 Ok(Ok(())) => {
209 tracing::info!(%user_id, path = %user_git_dir.display(), "deleted git repos from disk");
210 }
211 Ok(Err(e)) => {
212 tracing::warn!(error = ?e, %user_id, path = %user_git_dir.display(), "failed to delete git repos from disk");
213 }
214 Err(e) => {
215 tracing::warn!(error = ?e, %user_id, path = %user_git_dir.display(), "git repo cleanup task panicked");
216 }
217 }
218 }
219 }
220
221 /// Delete accounts that were terminated >30 days ago.
222 ///
223 /// Per-user S3 sweeps + git-repo removal + CASCADE delete are spawned off the
224 /// scheduler's advisory-lock-held tick so a backlog of expired accounts
225 /// (5 creators × 20 projects of multi-page `delete_prefix` calls) doesn't
226 /// extend the lock hold past `TICK_DURATION_ALERT_SECS`. Each step in
227 /// `cleanup_user_s3_and_delete` is idempotent: `delete_prefix` on a missing
228 /// prefix is a no-op, the CASCADE `delete_user` on a non-existent row affects
229 /// zero rows. If a second tick fires before all spawned cleanups complete
230 /// (extremely large accounts), the next `get_expired_terminated_ids` query
231 /// excludes already-deleted rows; the worst-case race is one extra harmless
232 /// no-op call against an in-flight target.
233 #[tracing::instrument(skip_all, name = "scheduler::delete_expired_terminated_accounts")]
234 pub(super) async fn delete_expired_terminated_accounts(state: &AppState) {
235 spawn_expired_account_cleanups(
236 state,
237 db::users::get_expired_terminated_ids(&state.db).await,
238 "termination_expired",
239 "terminated account",
240 "terminated_account_cleanup",
241 )
242 .await;
243 }
244
245 /// Delete creator accounts whose 90-day content removal grace period has expired.
246 /// Off-lock spawn pattern, see `delete_expired_terminated_accounts`.
247 #[tracing::instrument(skip_all, name = "scheduler::delete_expired_content_removal_accounts")]
248 pub(super) async fn delete_expired_content_removal_accounts(state: &AppState) {
249 spawn_expired_account_cleanups(
250 state,
251 db::users::get_expired_content_removal_ids(&state.db).await,
252 "content_removal_expired",
253 "content removal",
254 "content_removal_cleanup",
255 )
256 .await;
257 }
258
259 async fn spawn_expired_account_cleanups(
260 state: &AppState,
261 fetched: Result<Vec<db::UserId>, crate::error::AppError>,
262 event: &'static str,
263 label: &'static str,
264 job_name: &'static str,
265 ) {
266 let expired_ids = match fetched {
267 Ok(ids) if ids.is_empty() => {
268 let _ = db::scheduler_jobs::record_job_run(&state.db, job_name, 0).await;
269 return;
270 }
271 Ok(ids) => ids,
272 Err(e) => {
273 tracing::error!(error = ?e, %job_name, "failed to query expired accounts");
274 return;
275 }
276 };
277
278 let scheduled = expired_ids.len() as i64;
279
280 // Run the per-user sweeps in a SINGLE supervisor task spawned off the tick,
281 // draining through the shared bounded-concurrency helper. This keeps the
282 // off-lock property the doc comment relies on (the scheduler tick returns
283 // immediately, never holding the advisory lock across the sweeps) AND caps
284 // how many sweeps run at once. A bare `tokio::spawn` per expired account had
285 // no cap: a mass expiry (terminated-account or 90-day content-removal purge)
286 // could fan out N concurrent tasks, each holding several pool connections
287 // plus an S3 delete-prefix, and drain the connection pool out from under
288 // request handlers (Run #11 SERIOUS).
289 let supervisor_state = state.clone();
290 tokio::spawn(async move {
291 drain_cleanup_bounded(&supervisor_state, expired_ids, event, label).await;
292 });
293
294 // Record the count *scheduled*, not *deleted*: the deletes finish out of band.
295 // Operators read this metric to confirm the scheduler saw the work, not to
296 // count completed sweeps. cleanup_user_s3_and_delete logs its own success/
297 // failure per user.
298 let _ = db::scheduler_jobs::record_job_run(&state.db, job_name, scheduled).await;
299 }
300
301 /// Delete pending transactions older than 25 hours and release promo code reservations.
302 #[tracing::instrument(skip_all, name = "scheduler::cleanup_stale_pending_transactions")]
303 pub(super) async fn cleanup_stale_pending_transactions(state: &AppState) {
304 let promo_ids =
305 match db::transactions::cleanup_stale_pending(&state.db, chrono::Duration::hours(25)).await
306 {
307 Ok(ids) => ids,
308 Err(e) => {
309 tracing::error!(error = ?e, "failed to clean up stale pending transactions");
310 return;
311 }
312 };
313
314 // Cart checkouts produce N pending-tx rows that share a promo_code_id;
315 // release once per reservation, not once per row.
316 let unique_promo_ids: std::collections::HashSet<_> = promo_ids.into_iter().flatten().collect();
317
318 let mut released = 0i64;
319 for pc_id in unique_promo_ids {
320 if let Err(e) = db::promo_codes::release_use_count(&state.db, pc_id).await {
321 tracing::warn!(promo_code_id = %pc_id, error = ?e, "failed to release promo code use count");
322 } else {
323 released += 1;
324 }
325 }
326
327 if released > 0 {
328 tracing::info!(
329 released,
330 "released promo code reservations from stale pending transactions"
331 );
332 }
333 let _ = db::scheduler_jobs::record_job_run(&state.db, "stale_pending_cleanup", released).await;
334 }
335
336 /// NULL out IP addresses older than 30 days in user_sessions.
337 #[tracing::instrument(skip_all, name = "scheduler::scrub_stale_ip_addresses")]
338 pub(super) async fn scrub_stale_ip_addresses(state: &AppState) {
339 let cutoff = chrono::Utc::now() - chrono::Duration::days(30);
340
341 match sqlx::query(
342 "UPDATE user_sessions SET ip_address = NULL WHERE ip_address IS NOT NULL AND created_at < $1",
343 )
344 .bind(cutoff)
345 .execute(&state.db)
346 .await
347 {
348 Ok(r) => {
349 if r.rows_affected() > 0 {
350 tracing::info!(scrubbed = r.rows_affected(), "scrubbed stale IP addresses (30-day retention)");
351 }
352 let _ = db::scheduler_jobs::record_job_run(&state.db, "ip_scrub", r.rows_affected() as i64).await;
353 }
354 Err(e) => tracing::error!(error = ?e, "failed to scrub IPs from user_sessions"),
355 }
356 }
357
358 /// Permanently delete items that were soft-deleted more than 7 days ago.
359 /// Cleans up S3 objects (item files + version files) and decrements storage
360 /// before DB deletion to prevent orphaned storage and accounting drift.
361 ///
362 /// Accepted residual: this gathers
363 /// all expired-item S3 keys into one Vec before the batch delete. It is a daily
364 /// cron off the request path; steady-state (7-day window) is small, so the only
365 /// large allocation is transient, after a rare mass-delete event. Bounding it
366 /// per-tick means coupling a LIMIT on the key-gather to the same slice as the
367 /// CASCADE delete; deferred as low-value for a cron path. Revisit if it matters.
368 #[tracing::instrument(skip_all, name = "scheduler::purge_expired_deleted_items")]
369 pub(super) async fn purge_expired_deleted_items(state: &AppState) {
370 // Collect S3 keys from items AND their versions before CASCADE delete destroys the data
371 let mut all_s3_keys: Vec<(String, String)> = Vec::new();
372
373 // Item keys are audio/video (private bucket) mixed with the item cover
374 // (public bucket); enqueue each under both so the reaper hits the right one
375 // (see `both_bucket_delete`).
376 match db::items::get_expired_deleted_item_s3_keys(&state.db).await {
377 Ok(keys) => {
378 for key in &keys {
379 all_s3_keys.extend(crate::storage::both_bucket_delete(key));
380 }
381 }
382 Err(e) => {
383 tracing::error!(error = ?e, "failed to query item S3 keys for items pending purge");
384 }
385 }
386
387 match db::items::get_expired_deleted_item_version_s3_keys(&state.db).await {
388 Ok(keys) => {
389 for key in &keys {
390 // Version downloads are gated media, always the private bucket.
391 all_s3_keys.push((
392 key.clone(),
393 crate::storage::S3Bucket::Main.as_str().to_string(),
394 ));
395 }
396 }
397 Err(e) => {
398 tracing::error!(error = ?e, "failed to query version S3 keys for items pending purge");
399 }
400 }
401
402 // Gallery images (item_images) cascade away with the purged items; collect
403 // their keys too or the objects orphan with no durable record (Run #18 B2).
404 // CDN-served → public bucket post-promote (or staging in main); enqueue both.
405 match db::gallery_images::s3_keys_for_expired_purged_items(&state.db).await {
406 Ok(keys) => {
407 for key in &keys {
408 all_s3_keys.extend(crate::storage::both_bucket_delete(key));
409 }
410 }
411 Err(e) => {
412 tracing::error!(error = ?e, "failed to query gallery S3 keys for items pending purge");
413 }
414 }
415
416 // Enqueue all keys as a durable safety net before any destructive work
417 if !all_s3_keys.is_empty()
418 && let Err(e) = db::pending_s3_deletions::enqueue_deletions(
419 &state.db,
420 &all_s3_keys,
421 "purge_deleted_items",
422 )
423 .await
424 {
425 tracing::error!(error = ?e, "failed to enqueue S3 deletions for purged items, aborting purge");
426 return;
427 }
428
429 if let Some(ref s3) = state.storage.s3
430 && !all_s3_keys.is_empty()
431 {
432 let keys_only: Vec<crate::storage::S3Key> = all_s3_keys
433 .iter()
434 .map(|(k, _)| crate::storage::S3Key::from_stored(k))
435 .collect();
436 if let Err(e) = s3
437 .delete_objects(&S3DeleteAuthority::new(), &keys_only)
438 .await
439 {
440 tracing::warn!(error = ?e, "batch S3 delete failed for purged items; pending_s3_deletions queue will retry");
441 }
442 tracing::info!(
443 count = all_s3_keys.len(),
444 "deleted S3 objects for purged items"
445 );
446 }
447
448 // Decrement each affected user's storage AND purge the items in ONE
449 // transaction. Previously these were two pool calls: a crash between them
450 // left the items un-purged but never re-measured, so the next tick decremented
451 // the SAME items again → `storage_used_bytes` under-counted (fuzz-2026-07-06
452 // LOW). Folding them means a crash rolls both back and the next tick redoes the
453 // pair cleanly. The measure and the DELETE share the transaction's fixed
454 // `NOW()`, so both act on exactly the same expired set.
455 let mut tx = match state.db.begin().await {
456 Ok(tx) => tx,
457 Err(e) => {
458 tracing::error!(error = ?e, "failed to open tx for item purge; will retry next tick");
459 return;
460 }
461 };
462
463 match db::items::get_expired_deleted_item_storage_by_user(&mut *tx).await {
464 Ok(user_sizes) => {
465 for (user_id, total_bytes) in &user_sizes {
466 if *total_bytes > 0
467 && let Err(e) =
468 db::creator_tiers::decrement_storage_used(&mut *tx, *user_id, *total_bytes)
469 .await
470 {
471 // A failed decrement poisons the tx; abort the whole purge so
472 // items are never deleted without their storage being credited
473 // back. The next tick retries the pair atomically.
474 tracing::warn!(user_id = %user_id, bytes = total_bytes, error = ?e,
475 "failed to decrement storage for purged items; aborting purge tx");
476 return;
477 }
478 }
479 }
480 Err(e) => {
481 tracing::error!(error = ?e, "failed to query storage sizes for items pending purge; aborting purge tx");
482 return;
483 }
484 }
485
486 let purged = match db::items::purge_expired_deleted_items(&mut *tx).await {
487 Ok(n) => n,
488 Err(e) => {
489 tracing::error!(error = ?e, "failed to purge expired soft-deleted items; aborting purge tx");
490 return;
491 }
492 };
493
494 if let Err(e) = tx.commit().await {
495 tracing::error!(error = ?e, "failed to commit item purge tx; will retry next tick");
496 return;
497 }
498
499 if purged > 0 {
500 tracing::info!(deleted = purged, "purged expired soft-deleted items");
501 }
502 let _ = db::scheduler_jobs::record_job_run(&state.db, "soft_delete_purge", purged as i64).await;
503 }
504
505 /// Outcome of a guarded single-key orphan S3 delete.
506 enum GuardedDelete {
507 /// Object deleted (S3 returned Ok).
508 Deleted,
509 /// Skipped: a live DB row still references the key (delete-then-reupload
510 /// race). The caller should clear its queue/record without deleting.
511 SkippedLive,
512 /// S3 delete failed (already logged). The caller decides retry vs. clear.
513 Failed,
514 }
515
516 /// The single funnel for deleting one non-prefix orphan S3 object.
517 ///
518 /// Performs the `is_s3_key_live` check FIRST, so an object that a fresh upload
519 /// reclaimed under the same key is never torpedoed. Both the pending-deletions
520 /// retry worker and the orphaned-upload reaper route every single-key delete
521 /// through here, neither can delete a key without the live-check. A sibling
522 /// that keeps the guard inline instead can delete a now-live deterministic-key
523 /// object.
524 async fn delete_orphan_key_guarded(
525 db: &sqlx::PgPool,
526 s3: &dyn crate::storage::StorageBackend,
527 bucket: &str,
528 s3_key: &str,
529 ) -> GuardedDelete {
530 match db::pending_s3_deletions::is_s3_key_live(db, bucket, s3_key).await {
531 Ok(true) => {
532 tracing::info!(s3_key = %s3_key, bucket = %bucket,
533 "orphan S3 delete skipped, key reclaimed by a live row (delete-then-reupload)");
534 return GuardedDelete::SkippedLive;
535 }
536 Ok(false) => {}
537 Err(e) => {
538 tracing::warn!(s3_key = %s3_key, error = ?e,
539 "live-key check failed; proceeding with delete attempt");
540 }
541 }
542 match s3
543 .delete_object(
544 &S3DeleteAuthority::new(),
545 &crate::storage::S3Key::from_stored(s3_key),
546 )
547 .await
548 {
549 Ok(()) => GuardedDelete::Deleted,
550 Err(e) => {
551 tracing::warn!(s3_key = %s3_key, bucket = %bucket, error = ?e, "orphan S3 delete failed");
552 GuardedDelete::Failed
553 }
554 }
555 }
556
557 /// Delete S3 objects from presigned uploads that were never confirmed (>24h old).
558 #[tracing::instrument(skip_all, name = "scheduler::cleanup_orphaned_uploads")]
559 pub(super) async fn cleanup_orphaned_uploads(state: &AppState) {
560 let stale = match db::pending_uploads::get_stale_pending_uploads(
561 &state.db,
562 chrono::Duration::hours(24),
563 )
564 .await
565 {
566 Ok(rows) if rows.is_empty() => {
567 let _ =
568 db::scheduler_jobs::record_job_run(&state.db, "orphaned_upload_cleanup", 0).await;
569 return;
570 }
571 Ok(rows) => rows,
572 Err(e) => {
573 tracing::error!(error = ?e, "failed to query stale pending uploads");
574 return;
575 }
576 };
577
578 let mut cleaned = 0i64;
579 let mut aborted = 0i64;
580 // Carry (key, bucket) pairs: the deletion is bucket-scoped so a key present
581 // in two buckets only clears the record for the bucket we reaped (B7).
582 let mut keys_to_delete: Vec<(String, String)> = Vec::with_capacity(stale.len());
583 // Transient S3 delete failures handed off to the durable deletion queue
584 // (which has a retry + dead-letter backstop) instead of dropping the only
585 // tracking row, clearing it on a transient failure permanently leaked the
586 // object (Run #2 Storage SERIOUS; the sibling retry worker already routes
587 // failures into this ladder).
588 let mut failed_keys: Vec<(String, String)> = Vec::new();
589
590 for (s3_key, bucket) in &stale {
591 let s3_client = match crate::storage::S3Bucket::from_db_str(bucket) {
592 crate::storage::S3Bucket::Synckit => state.storage.synckit_s3.as_ref(),
593 crate::storage::S3Bucket::Public => state.storage.public_s3.as_ref(),
594 crate::storage::S3Bucket::Main => state.storage.s3.as_ref(),
595 };
596 if let Some(s3) = s3_client {
597 // A multipart session that was started and never completed leaves NO
598 // object to delete, only uploaded parts that S3 bills for until they
599 // are aborted, so the delete below is a no-op against it and the
600 // parts would leak forever. Abort first, keyed off S3's own list so a
601 // session whose upload_id was never recorded is still caught.
602 //
603 // Best-effort: a failed abort must not block the object delete or
604 // strand the tracking row. The bucket lifecycle rule
605 // (AbortIncompleteMultipartUpload) is the backstop for anything
606 // missed here.
607 aborted += abort_orphan_multipart_sessions(s3.as_ref(), s3_key).await;
608
609 // Route through the guarded funnel: if a confirm already reclaimed
610 // this deterministic key, the live row owns the object and we must
611 // NOT delete it, just clear the stale pending_uploads record.
612 match delete_orphan_key_guarded(&state.db, s3.as_ref(), bucket, s3_key).await {
613 GuardedDelete::Deleted => {
614 cleaned += 1;
615 keys_to_delete.push((s3_key.clone(), bucket.clone()));
616 }
617 // Live row owns the key now: clear the stale record, keep object.
618 GuardedDelete::SkippedLive => keys_to_delete.push((s3_key.clone(), bucket.clone())),
619 // Delete failed (transient): hand off to the durable deletion
620 // queue rather than dropping the tracking row and leaking.
621 GuardedDelete::Failed => failed_keys.push((s3_key.clone(), bucket.clone())),
622 }
623 } else {
624 // S3 not configured for this bucket; remove the DB record anyway
625 keys_to_delete.push((s3_key.clone(), bucket.clone()));
626 }
627 }
628
629 // Enqueue transient failures into the durable deletion queue BEFORE clearing
630 // their tracking rows, so a key is never dropped by both. If the handoff
631 // itself fails, leave the pending_uploads rows for the next reaper tick
632 // rather than leaking.
633 if !failed_keys.is_empty() {
634 match db::pending_s3_deletions::enqueue_deletions(
635 &state.db,
636 &failed_keys,
637 "orphan-reaper-retry",
638 )
639 .await
640 {
641 Ok(()) => {
642 tracing::warn!(
643 count = failed_keys.len(),
644 "orphan reaper: S3 delete failed; handed off to durable deletion queue for retry"
645 );
646 keys_to_delete.extend(failed_keys);
647 }
648 Err(e) => {
649 tracing::error!(
650 error = ?e, count = failed_keys.len(),
651 "orphan reaper: could not enqueue transient failures to deletion queue; leaving tracking rows for next tick"
652 );
653 }
654 }
655 }
656
657 if !keys_to_delete.is_empty()
658 && let Err(e) =
659 db::pending_uploads::delete_pending_uploads(&state.db, &keys_to_delete).await
660 {
661 tracing::error!(error = ?e, "failed to delete pending upload records");
662 }
663
664 if cleaned > 0 || aborted > 0 {
665 tracing::info!(cleaned, aborted, "cleaned up orphaned presigned uploads");
666 }
667 let _ = db::scheduler_jobs::record_job_run(&state.db, "orphaned_upload_cleanup", cleaned).await;
668 }
669
670 /// Abort every in-progress multipart session for `s3_key`, returning how many
671 /// were aborted.
672 ///
673 /// Best-effort by design: the caller still deletes the object and clears the
674 /// tracking row regardless, and the bucket's `AbortIncompleteMultipartUpload`
675 /// lifecycle rule catches whatever fails here. Returning a count rather than a
676 /// Result keeps a storage hiccup from stranding the rest of the reap.
677 /// Public via a single re-export on `scheduler` (the module itself stays
678 /// private) so integration tests can drive it against a test backend without a
679 /// database or an `AppState`.
680 pub async fn abort_orphan_multipart_sessions(
681 s3: &dyn crate::storage::StorageBackend,
682 s3_key: &str,
683 ) -> i64 {
684 let upload_ids = match s3.list_multipart_uploads_for_key(s3_key).await {
685 Ok(ids) => ids,
686 Err(e) => {
687 tracing::warn!(
688 s3_key = %s3_key, error = ?e,
689 "orphan reaper: could not list multipart sessions; lifecycle rule is the backstop"
690 );
691 return 0;
692 }
693 };
694
695 let key = crate::storage::S3Key::from_stored(s3_key);
696 let mut aborted = 0i64;
697 for upload_id in upload_ids {
698 match s3.abort_multipart_upload(&key, &upload_id).await {
699 Ok(()) => {
700 aborted += 1;
701 tracing::info!(
702 s3_key = %s3_key, %upload_id,
703 "orphan reaper: aborted an abandoned multipart session"
704 );
705 }
706 Err(e) => tracing::warn!(
707 s3_key = %s3_key, %upload_id, error = ?e,
708 "orphan reaper: failed to abort multipart session; lifecycle rule is the backstop"
709 ),
710 }
711 }
712 aborted
713 }
714
715 /// Remove stale cart items (>30 days old) and items that became unavailable.
716 #[tracing::instrument(skip_all, name = "scheduler::cleanup_cart_items")]
717 pub(super) async fn cleanup_cart_items(state: &AppState) {
718 match db::cart::cleanup_stale_cart_items(&state.db, chrono::Duration::days(30)).await {
719 Ok(n) if n > 0 => tracing::info!(removed = n, "cleaned up stale cart items"),
720 Err(e) => tracing::error!(error = ?e, "failed to clean up stale cart items"),
721 _ => {}
722 }
723 match db::cart::cleanup_unavailable_cart_items(&state.db).await {
724 Ok(n) if n > 0 => tracing::info!(removed = n, "cleaned up unavailable cart items"),
725 Err(e) => tracing::error!(error = ?e, "failed to clean up unavailable cart items"),
726 _ => {}
727 }
728 }
729
730 /// Retry stale pending S3 deletions (older than 10 minutes, batch of 100).
731 #[tracing::instrument(skip_all, name = "scheduler::retry_pending_s3_deletions")]
732 pub(super) async fn retry_pending_s3_deletions(state: &AppState) {
733 let stale = match db::pending_s3_deletions::get_stale_pending(
734 &state.db,
735 chrono::Duration::minutes(10),
736 100,
737 )
738 .await
739 {
740 Ok(rows) => rows,
741 Err(e) => {
742 tracing::error!(error = ?e, "failed to fetch stale pending S3 deletions");
743 return;
744 }
745 };
746
747 if stale.is_empty() {
748 let _ = db::scheduler_jobs::record_job_run(&state.db, "s3_deletion_retry", 0).await;
749 return;
750 }
751
752 let mut completed_ids = Vec::new();
753 let mut dead_letter_ids = Vec::new();
754 for row in &stale {
755 if row.attempts >= 10 {
756 tracing::error!(s3_key = %row.s3_key, bucket = %row.bucket, source = %row.source, attempts = row.attempts,
757 "S3 deletion dead-lettered after 10 attempts, moving to dead-letter table for manual triage");
758 dead_letter_ids.push(row.id);
759 continue;
760 } else if row.attempts >= 5 {
761 tracing::warn!(s3_key = %row.s3_key, bucket = %row.bucket, source = %row.source, attempts = row.attempts,
762 "S3 deletion stuck after 5+ attempts");
763 }
764
765 let s3 = match crate::storage::S3Bucket::from_db_str(&row.bucket) {
766 crate::storage::S3Bucket::Synckit => state.storage.synckit_s3.as_ref(),
767 crate::storage::S3Bucket::Public => state.storage.public_s3.as_ref(),
768 crate::storage::S3Bucket::Main => state.storage.s3.as_ref(),
769 };
770
771 if let Some(s3) = s3 {
772 if row.s3_key.ends_with('/') {
773 // Prefix delete (account-cascade cleanup). Bypasses the per-key
774 // live-check by design, but carries a liveness guard symmetric
775 // across buckets: a `{user_id}/` (main) or `{app_id}/` /
776 // `ota/{app_id}/` (synckit) prefix wipes an entire creator's or
777 // app's storage and is only ever enqueued by cleanup that deletes
778 // the owning row in the same pass. A STILL-LIVE owner means
779 // something wrongly enqueued a wipe, refuse and let the row climb
780 // toward dead-letter triage rather than nuking live files. The
781 // synckit branch was previously unguarded (ultra-fuzz Run 11 Storage LOW).
782 let live_owner: Option<String> =
783 match crate::storage::S3Bucket::from_db_str(&row.bucket) {
784 // Both the main and public buckets key content under
785 // `{user_id}/`, so the owner guard is identical.
786 crate::storage::S3Bucket::Main | crate::storage::S3Bucket::Public => {
787 match row
788 .s3_key
789 .strip_suffix('/')
790 .and_then(|s| s.parse::<db::UserId>().ok())
791 {
792 Some(uid)
793 if matches!(
794 db::users::get_user_by_id(&state.db, uid).await,
795 Ok(Some(_))
796 ) =>
797 {
798 Some(format!("user {uid}"))
799 }
800 _ => None,
801 }
802 }
803 crate::storage::S3Bucket::Synckit => {
804 let app_seg = row.s3_key.strip_prefix("ota/").unwrap_or(&row.s3_key);
805 match app_seg
806 .strip_suffix('/')
807 .and_then(|s| s.parse::<db::SyncAppId>().ok())
808 {
809 Some(app_id)
810 if matches!(
811 db::synckit::get_sync_app_by_id(&state.db, app_id).await,
812 Ok(Some(_))
813 ) =>
814 {
815 Some(format!("sync app {app_id}"))
816 }
817 _ => None,
818 }
819 }
820 };
821 if let Some(owner) = live_owner {
822 tracing::error!(s3_key = %row.s3_key, %owner,
823 "refusing prefix S3 delete: owner still exists, parking for dead-letter triage instead of wiping live storage");
824 continue;
825 }
826 match s3
827 .delete_prefix(&S3DeleteAuthority::new(), &row.s3_key)
828 .await
829 {
830 Ok(()) => completed_ids.push(row.id),
831 Err(e) => {
832 tracing::warn!(s3_key = %row.s3_key, error = ?e, "retry S3 prefix deletion failed");
833 }
834 }
835 } else {
836 // Single-key delete, routed through the one guarded funnel so the
837 // delete-then-reupload live-check can never be skipped. The same
838 // funnel backs the orphaned-upload reaper.
839 match delete_orphan_key_guarded(&state.db, s3.as_ref(), &row.bucket, &row.s3_key)
840 .await
841 {
842 GuardedDelete::Deleted | GuardedDelete::SkippedLive => {
843 completed_ids.push(row.id);
844 }
845 GuardedDelete::Failed => {} // leave queued; climbs toward dead-letter
846 }
847 }
848 } else {
849 // S3 not configured, remove from queue (can't delete what doesn't exist)
850 completed_ids.push(row.id);
851 }
852 }
853
854 // Move permanently-failing rows to the dead-letter table (durable, operator-
855 // visible) rather than silently DELETEing them and orphaning the S3 object.
856 if !dead_letter_ids.is_empty() {
857 match db::pending_s3_deletions::move_to_dead_letter(&state.db, &dead_letter_ids).await {
858 Ok(moved) => tracing::warn!(
859 moved,
860 "moved permanently-failing S3 deletions to dead-letter table, manual triage required"
861 ),
862 Err(e) => {
863 tracing::error!(error = ?e, "failed to move S3 deletions to dead-letter table");
864 }
865 }
866 }
867
868 if !completed_ids.is_empty() {
869 if let Err(e) = db::pending_s3_deletions::remove_completed(&state.db, &completed_ids).await
870 {
871 tracing::error!(error = ?e, "failed to dequeue completed S3 deletions");
872 } else {
873 tracing::info!(
874 completed = completed_ids.len(),
875 total = stale.len(),
876 "retried pending S3 deletions"
877 );
878 }
879 }
880
881 if !completed_ids.is_empty() || !dead_letter_ids.is_empty() {
882 let processed = (completed_ids.len() + dead_letter_ids.len()) as i64;
883 let _ = db::scheduler_jobs::record_job_run(&state.db, "s3_deletion_retry", processed).await;
884 }
885 }
886
887 /// Test-only synchronous drain of the pending-S3-deletion queue (`main` bucket).
888 ///
889 /// Run [`cleanup_orphaned_uploads`] once, synchronously.
890 ///
891 /// The reaper is `pub(super)` and the scheduler drives it on a tick, so no
892 /// integration test could reach it, including the branch that hands a failed S3
893 /// delete to the durable deletion queue instead of dropping the tracking row,
894 /// which is where an object leaks. Exposed via
895 /// `TestHarness::run_orphan_upload_reaper`.
896 #[doc(hidden)]
897 #[tracing::instrument(skip_all, name = "scheduler::cleanup_orphaned_uploads_for_test")]
898 pub async fn cleanup_orphaned_uploads_for_test(state: &AppState) {
899 cleanup_orphaned_uploads(state).await;
900 }
901
902 /// The S3 delete a confirm/delete handler triggers is asynchronous: handlers
903 /// only [`enqueue_s3_orphan`](crate::routes::storage::enqueue_s3_orphan), and
904 /// the scheduler's [`retry_pending_s3_deletions`] performs the actual delete
905 /// later. Integration tests can't mint [`S3DeleteAuthority`] and don't run the
906 /// scheduler, so a test that asserts an object is gone must force the queued
907 /// deletes first. This runs them immediately against the given pool + storage,
908 /// mirroring the scheduler's per-row guarded-delete path. Returns the number of
909 /// objects actually deleted. Exposed via `TestHarness::drain_s3_deletions`.
910 #[doc(hidden)]
911 #[tracing::instrument(skip_all, name = "scheduler::drain_pending_s3_deletions_for_test")]
912 pub async fn drain_pending_s3_deletions_for_test(
913 pool: &sqlx::PgPool,
914 s3: &dyn crate::storage::StorageBackend,
915 ) -> usize {
916 let Ok(stale) =
917 db::pending_s3_deletions::get_stale_pending(pool, chrono::Duration::zero(), 1000).await
918 else {
919 return 0;
920 };
921 let mut completed = Vec::new();
922 let mut deleted = 0usize;
923 for row in &stale {
924 // Storage tests only exercise the main bucket; synckit needs its own s3.
925 if crate::storage::S3Bucket::from_db_str(&row.bucket) != crate::storage::S3Bucket::Main {
926 continue;
927 }
928 match delete_orphan_key_guarded(pool, s3, &row.bucket, &row.s3_key).await {
929 GuardedDelete::Deleted => {
930 deleted += 1;
931 completed.push(row.id);
932 }
933 GuardedDelete::SkippedLive => completed.push(row.id),
934 GuardedDelete::Failed => {}
935 }
936 }
937 if !completed.is_empty() {
938 let _ = db::pending_s3_deletions::remove_completed(pool, &completed).await;
939 }
940 deleted
941 }
942
943 #[cfg(test)]
944 mod tests {
945 use super::*;
946
947 #[tokio::test]
948 async fn cleanup_git_repos_removes_directory() {
949 let tmp = tempfile::tempdir().unwrap();
950 let git_root = tmp.path();
951 let user_dir = git_root.join("testuser");
952 std::fs::create_dir_all(user_dir.join("repo.git")).unwrap();
953 std::fs::write(user_dir.join("repo.git/HEAD"), "ref: refs/heads/main\n").unwrap();
954
955 let user_id = db::UserId::nil();
956 cleanup_git_repos_on_disk(git_root.to_str().unwrap(), "testuser", user_id).await;
957
958 assert!(!user_dir.exists(), "user git directory should be deleted");
959 }
960
961 #[tokio::test]
962 async fn cleanup_git_repos_noop_if_missing() {
963 let tmp = tempfile::tempdir().unwrap();
964 let user_id = db::UserId::nil();
965 cleanup_git_repos_on_disk(tmp.path().to_str().unwrap(), "nonexistent", user_id).await;
966 }
967 }
968