Skip to main content

max / makenotwork

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