From 5b7f83a6e3b41d06415fa8eb5c4eb767dcf12a39 Mon Sep 17 00:00:00 2001 From: DarkSky <25152247+darkskygit@users.noreply.github.com> Date: Thu, 2 Jul 2026 06:14:01 +0800 Subject: [PATCH] feat(server): batch blob gc (#15183) --- packages/backend/native/Cargo.toml | 2 +- .../runtime/storage_runtime/blob_cleanup.rs | 119 +++++++++-- .../native/src/runtime/storage_runtime/mod.rs | 92 ++++++++- .../storage_runtime/object_storage/client.rs | 188 +++++++++++++++++- .../storage_runtime/object_storage/error.rs | 8 + .../storage_runtime/object_storage/types.rs | 6 + .../core/storage/__tests__/blob-job.spec.ts | 132 ++++++++++++ .../server/src/core/storage/blob-job.ts | 122 ++++++++++++ 8 files changed, 639 insertions(+), 30 deletions(-) diff --git a/packages/backend/native/Cargo.toml b/packages/backend/native/Cargo.toml index b45056268..40fa7f950 100644 --- a/packages/backend/native/Cargo.toml +++ b/packages/backend/native/Cargo.toml @@ -64,7 +64,7 @@ sqlx = { workspace = true, default-features = false, features = [ ] } thiserror.workspace = true tiktoken-rs = { workspace = true } -tokio = { workspace = true, features = ["rt-multi-thread", "sync"] } +tokio = { workspace = true, features = ["rt-multi-thread", "sync", "time"] } url = { workspace = true } uuid = { workspace = true, features = ["v4"] } v_htmlescape = { workspace = true } diff --git a/packages/backend/native/src/runtime/storage_runtime/blob_cleanup.rs b/packages/backend/native/src/runtime/storage_runtime/blob_cleanup.rs index c58274115..808d0f364 100644 --- a/packages/backend/native/src/runtime/storage_runtime/blob_cleanup.rs +++ b/packages/backend/native/src/runtime/storage_runtime/blob_cleanup.rs @@ -1,3 +1,5 @@ +use std::collections::HashMap; + use chrono::{DateTime, Duration, Utc}; use sqlx::{FromRow, PgPool}; @@ -19,6 +21,12 @@ struct MarkedCandidateRow { blob_key: String, } +struct DeletableCandidate { + workspace_id: String, + blob_key: String, + object_key: String, +} + fn push_workspace_once(workspace_ids: &mut Vec, workspace_id: &str) { if !workspace_ids.iter().any(|id| id == workspace_id) { workspace_ids.push(workspace_id.to_string()); @@ -488,6 +496,7 @@ impl StorageRuntime { failed: 0, workspace_ids: Vec::new(), }; + let mut deletable_candidates = Vec::new(); for row in rows { if projection_is_stale(&pool, &row.workspace_id).await? @@ -509,7 +518,6 @@ impl StorageRuntime { } let object_key = format!("{}/{}", row.workspace_id, row.blob_key); - let mut object_was_missing = false; let metadata = match self.object_storage_head(object_key.clone()).await { Ok(metadata) => metadata, Err(err) => { @@ -544,23 +552,12 @@ impl StorageRuntime { .await?; continue; } - if let Err(err) = self.object_storage_delete(object_key).await { - result.failed += 1; - mark_candidate_status( - &pool, - &run_id, - &row.workspace_id, - &row.blob_key, - "failed", - serde_json::json!({ "failure": "object_delete_failed" }), - Some(&err.to_string()), - ) - .await?; - continue; - } - result.deleted_objects += 1; - } else { - object_was_missing = true; + deletable_candidates.push(DeletableCandidate { + workspace_id: row.workspace_id, + blob_key: row.blob_key, + object_key, + }); + continue; } let deleted_metadata = @@ -597,13 +594,97 @@ impl StorageRuntime { "executed", serde_json::json!({ "deletedMetadata": deleted_metadata, - "objectMissingBeforeDelete": object_was_missing, + "objectMissingBeforeDelete": true, }), None, ) .await?; } + if !deletable_candidates.is_empty() { + let object_keys = deletable_candidates + .iter() + .map(|candidate| candidate.object_key.clone()) + .collect::>(); + let outcomes = match self.object_storage_delete_many(object_keys.clone()).await { + Ok(outcomes) => outcomes, + Err(err) => object_keys + .into_iter() + .map(|key| super::object_storage::types::ObjectDeleteOutcome { + key, + error: Some(err.to_string()), + }) + .collect(), + }; + let mut outcomes_by_key = outcomes + .into_iter() + .map(|outcome| (outcome.key, outcome.error)) + .collect::>(); + + for row in deletable_candidates { + let delete_error = match outcomes_by_key.remove(&row.object_key) { + Some(Some(error)) => Some(error), + Some(None) => None, + None => Some("DeleteObjects response did not include this key".to_string()), + }; + if let Some(error) = delete_error { + result.failed += 1; + mark_candidate_status( + &pool, + &run_id, + &row.workspace_id, + &row.blob_key, + "failed", + serde_json::json!({ "failure": "object_delete_failed" }), + Some(&error), + ) + .await?; + continue; + } + result.deleted_objects += 1; + + let deleted_metadata = + match sqlx::query("DELETE FROM blobs WHERE workspace_id = $1 AND key = $2 AND deleted_at IS NULL") + .bind(&row.workspace_id) + .bind(&row.blob_key) + .execute(&pool) + .await + { + Ok(result) => result.rows_affected() as i64, + Err(err) => { + result.failed += 1; + mark_candidate_status( + &pool, + &run_id, + &row.workspace_id, + &row.blob_key, + "failed", + serde_json::json!({ "failure": "metadata_delete_failed" }), + Some(&err.to_string()), + ) + .await?; + continue; + } + }; + result.deleted_metadata += deleted_metadata; + push_workspace_once(&mut result.workspace_ids, &row.workspace_id); + + mark_candidate_status( + &pool, + &run_id, + &row.workspace_id, + &row.blob_key, + "executed", + serde_json::json!({ + "deletedMetadata": deleted_metadata, + "objectMissingBeforeDelete": false, + }), + None, + ) + .await?; + } + } + finish_execute_run(&pool, &run_id, &result).await?; Ok(result) } diff --git a/packages/backend/native/src/runtime/storage_runtime/mod.rs b/packages/backend/native/src/runtime/storage_runtime/mod.rs index 83e351ed4..403231518 100644 --- a/packages/backend/native/src/runtime/storage_runtime/mod.rs +++ b/packages/backend/native/src/runtime/storage_runtime/mod.rs @@ -12,7 +12,7 @@ use serde::Deserialize; use serde_json::{Map, Value}; use sha2::{Digest, Sha256}; use sqlx::{PgPool, Row, postgres::PgPoolOptions}; -use tokio::sync::Mutex; +use tokio::{sync::Mutex, task::JoinSet}; mod assetpack; mod blob_cleanup; @@ -23,7 +23,9 @@ pub(crate) mod object_storage; use self::object_storage::{ ObjectStorageConfig, StorageProviderConfig, - types::{ObjectGetResult, ObjectListEntry, ObjectMetadata, ObjectPutMetadata, checksum_crc32_base64}, + types::{ + ObjectDeleteOutcome, ObjectGetResult, ObjectListEntry, ObjectMetadata, ObjectPutMetadata, checksum_crc32_base64, + }, }; pub(super) use super::{ RuntimeError, RuntimeResult, @@ -38,6 +40,8 @@ pub(super) use super::{ }; const MAX_BLOB_SIZE: i64 = i32::MAX as i64; +const OBJECT_DELETE_MANY_CHUNK_SIZE: usize = 500; +const OBJECT_DELETE_MANY_CONCURRENCY: usize = 3; type Result = RuntimeResult; @@ -680,6 +684,15 @@ impl StorageRuntime { } } + pub(crate) async fn object_storage_delete_many(&self, keys: Vec) -> Result> { + let backend = self.backend_for_scope("blob")?; + match backend { + StorageBackendConfig::Fs(config) => Ok(delete_many_fs(config, keys)), + StorageBackendConfig::Assetpack(config) => delete_many_assetpack(config, keys).await, + StorageBackendConfig::S3(config) => delete_many_s3(config, keys).await, + } + } + pub(crate) async fn object_storage_abort_upload(&self, key: &str, upload_id: &str) -> Result<()> { match self.backend_for_scope("blob")? { StorageBackendConfig::Fs(_) | StorageBackendConfig::Assetpack(_) => Ok(()), @@ -752,10 +765,6 @@ impl StorageRuntime { Ok(metadata.map(Into::into)) } - pub(crate) async fn object_storage_delete(&self, key: String) -> Result<()> { - self.object_storage_delete_object(&key).await - } - async fn complete_fs_workspace_blob( &self, config: FsStorageConfig, @@ -1205,6 +1214,77 @@ fn fs_delete(config: &FsStorageConfig, key: &str) -> Result<()> { Ok(()) } +fn delete_many_fs(config: FsStorageConfig, keys: Vec) -> Vec { + keys + .into_iter() + .map(|key| { + let error = fs_delete(&config, &key).err().map(|err| err.to_string()); + ObjectDeleteOutcome { key, error } + }) + .collect() +} + +async fn delete_many_assetpack(config: FsStorageConfig, keys: Vec) -> Result> { + let mut outcomes = Vec::with_capacity(keys.len()); + for key in keys { + let error = assetpack::delete(&config, "blob", &key) + .await + .err() + .map(|err| err.to_string()); + outcomes.push(ObjectDeleteOutcome { key, error }); + } + Ok(outcomes) +} + +async fn delete_many_s3(config: ObjectStorageConfig, keys: Vec) -> Result> { + let client = config.build_client()?; + let mut chunks = keys + .chunks(OBJECT_DELETE_MANY_CHUNK_SIZE) + .map(|chunk| chunk.to_vec()) + .collect::>() + .into_iter(); + let mut tasks = JoinSet::new(); + let mut outcomes = Vec::new(); + + for _ in 0..OBJECT_DELETE_MANY_CONCURRENCY { + let Some(chunk) = chunks.next() else { + break; + }; + let client = client.clone(); + tasks.spawn(async move { + let fallback = chunk.clone(); + let result = client.delete_many(chunk).await.map_err(RuntimeError::from); + (fallback, result) + }); + } + + while let Some(result) = tasks.join_next().await { + match result { + Ok((_chunk, Ok(batch_outcomes))) => outcomes.extend(batch_outcomes), + Ok((chunk, Err(err))) => outcomes.extend(chunk.into_iter().map(|key| ObjectDeleteOutcome { + key, + error: Some(err.to_string()), + })), + Err(err) => { + return Err(RuntimeError::invalid_state(format!( + "StorageRuntime delete batch task failed: {err}" + ))); + } + } + + if let Some(chunk) = chunks.next() { + let client = client.clone(); + tasks.spawn(async move { + let fallback = chunk.clone(); + let result = client.delete_many(chunk).await.map_err(RuntimeError::from); + (fallback, result) + }); + } + } + + Ok(outcomes) +} + fn read_fs_metadata(path: &Path) -> Result> { let raw = match fs::read_to_string(PathBuf::from(format!("{}.metadata.json", path.display()))) { Ok(raw) => raw, diff --git a/packages/backend/native/src/runtime/storage_runtime/object_storage/client.rs b/packages/backend/native/src/runtime/storage_runtime/object_storage/client.rs index cacfc3d43..e5697ca90 100644 --- a/packages/backend/native/src/runtime/storage_runtime/object_storage/client.rs +++ b/packages/backend/native/src/runtime/storage_runtime/object_storage/client.rs @@ -12,23 +12,29 @@ use rustls::RootCertStore; use rusty_s3::{ Bucket, Credentials, actions::{ - AbortMultipartUpload, CompleteMultipartUpload, CreateMultipartUpload, DeleteObject, GetObject, HeadObject, - ListObjectsV2, ListParts, PutObject, S3Action, UploadPart, + AbortMultipartUpload, CompleteMultipartUpload, CreateMultipartUpload, DeleteObject, DeleteObjects, + DeleteObjectsResponse, GetObject, HeadObject, ListObjectsV2, ListParts, ObjectIdentifier, PutObject, S3Action, + UploadPart, }, }; +use tokio::time::sleep; use url::Url; use super::{ error::{ObjectStorageError, ObjectStorageResult}, types::{ - MultipartUploadInitResult, MultipartUploadPart, ObjectGetResult, ObjectListEntry, ObjectListPage, ObjectMetadata, - ObjectPutMetadata, PresignedObjectRequest, completed_multipart_parts, trim_etag, + MultipartUploadInitResult, MultipartUploadPart, ObjectDeleteOutcome, ObjectGetResult, ObjectListEntry, + ObjectListPage, ObjectMetadata, ObjectPutMetadata, PresignedObjectRequest, completed_multipart_parts, trim_etag, }, }; const DEFAULT_REQUEST_TIMEOUT_MS: u64 = 30_000; const MAX_MULTIPART_PART_NUMBER: i32 = 10_000; const MAX_RESPONSE_BODY_BYTES: usize = i32::MAX as usize; +const DELETE_OBJECTS_MAX_KEYS: usize = 1000; +const DELETE_OBJECTS_MAX_ATTEMPTS: usize = 5; +const DELETE_OBJECTS_BACKOFF_BASE_MS: u64 = 1_000; +const DELETE_OBJECTS_BACKOFF_MAX_MS: u64 = 30_000; #[derive(Clone)] struct StorageHttpRequest { @@ -564,6 +570,99 @@ impl ObjectStorageClient { ensure_success_status(&response, &format!("ObjectStorage delete failed for {key}"))?; Ok(()) } + + pub(crate) async fn delete_many(&self, keys: Vec) -> ObjectStorageResult> { + if keys.is_empty() { + return Ok(Vec::new()); + } + if keys.len() > DELETE_OBJECTS_MAX_KEYS { + return Err(ObjectStorageError::InvalidInput(format!( + "DeleteObjects supports at most {DELETE_OBJECTS_MAX_KEYS} keys" + ))); + } + + let mut pending_keys = keys; + let mut outcomes = Vec::new(); + let mut last_error = None; + for attempt in 0..DELETE_OBJECTS_MAX_ATTEMPTS { + match self.delete_many_once(&pending_keys).await { + Ok(batch_outcomes) => { + let (retryable, completed) = split_delete_many_outcomes(batch_outcomes); + outcomes.extend(completed); + if retryable.is_empty() { + return Ok(outcomes); + } + if attempt + 1 >= DELETE_OBJECTS_MAX_ATTEMPTS { + outcomes.extend(retryable.into_iter().map(|(key, error)| ObjectDeleteOutcome { + key, + error: Some(error), + })); + return Ok(outcomes); + } + pending_keys = retryable.into_iter().map(|(key, _)| key).collect(); + sleep(Duration::from_millis(delete_objects_backoff_ms(attempt))).await; + } + Err(err) if err.is_retryable_http_status() && attempt + 1 < DELETE_OBJECTS_MAX_ATTEMPTS => { + last_error = Some(err); + sleep(Duration::from_millis(delete_objects_backoff_ms(attempt))).await; + } + Err(err) => return Err(err), + } + } + + Err(last_error.unwrap_or_else(|| { + ObjectStorageError::InvalidInput("DeleteObjects retry exhausted without an error".to_string()) + })) + } + + async fn delete_many_once(&self, keys: &[String]) -> ObjectStorageResult> { + let objects = keys + .iter() + .map(|key| ObjectIdentifier::new(key.clone())) + .collect::>(); + let mut action = DeleteObjects::new(&self.bucket, Some(&self.credentials), objects.iter()); + action.set_quiet(false); + let (body, content_md5) = action.clone().body_with_md5(); + action.headers_mut().insert("content-md5", content_md5.clone()); + action + .headers_mut() + .insert("content-type", "application/xml".to_string()); + action.headers_mut().insert("content-length", body.len().to_string()); + + let response = self + .http + .execute(StorageHttpRequest { + method: Method::POST, + url: action.sign(expires_in(self.presign_expires_in_seconds)), + headers: HashMap::from([ + ("content-md5".to_string(), content_md5), + ("content-type".to_string(), "application/xml".to_string()), + ("content-length".to_string(), body.len().to_string()), + ]), + body: Some(body.into_bytes()), + max_response_body_bytes: MAX_RESPONSE_BODY_BYTES, + }) + .await + .map_err(|source| operation_error("ObjectStorage delete many failed", source))?; + let body = ensure_success_text(response, "ObjectStorage delete many failed".to_string())?; + let parsed = DeleteObjectsResponse::parse(&body).map_err(|source| ObjectStorageError::InvalidXml { + context: "ObjectStorage parse delete many response failed".to_string(), + source, + })?; + let mut outcomes = keys + .iter() + .map(|key| ObjectDeleteOutcome { + key: key.clone(), + error: None, + }) + .collect::>(); + for error in parsed.errors { + if let Some(outcome) = outcomes.iter_mut().find(|outcome| outcome.key == error.key) { + outcome.error = Some(format!("{}: {}", error.code, error.message)); + } + } + Ok(outcomes) + } } fn insert_action_headers<'a, T: S3Action<'a>>(action: &mut T, headers: &HashMap) { @@ -665,6 +764,37 @@ fn complete_multipart_body(parts: &[MultipartUploadPart]) -> String { body } +fn delete_objects_backoff_ms(attempt: usize) -> u64 { + DELETE_OBJECTS_BACKOFF_BASE_MS + .saturating_mul(2_u64.saturating_pow(attempt as u32)) + .min(DELETE_OBJECTS_BACKOFF_MAX_MS) +} + +fn is_retryable_delete_objects_error(error: &str) -> bool { + error.starts_with("SlowDown:") + || error.starts_with("InternalError:") + || error.starts_with("ServiceUnavailable:") + || error.starts_with("RequestTimeout:") + || error.starts_with("Throttling:") + || error.starts_with("ThrottlingException:") + || error.starts_with("TooManyRequests:") +} + +fn split_delete_many_outcomes(outcomes: Vec) -> (Vec<(String, String)>, Vec) { + let mut retryable = Vec::new(); + let mut completed = Vec::new(); + for outcome in outcomes { + if let Some(error) = &outcome.error + && is_retryable_delete_objects_error(error) + { + retryable.push((outcome.key, error.clone())); + continue; + } + completed.push(outcome); + } + (retryable, completed) +} + fn xml_escape(value: &str) -> String { value.replace('&', "&").replace('<', "<").replace('>', ">") } @@ -734,6 +864,56 @@ mod tests { assert!(!is_not_found_body(b"AccessDenied")); } + #[test] + fn delete_objects_backoff_is_capped() { + assert_eq!(delete_objects_backoff_ms(0), 1_000); + assert_eq!(delete_objects_backoff_ms(1), 2_000); + assert_eq!(delete_objects_backoff_ms(10), 30_000); + } + + #[test] + fn delete_objects_outcomes_retry_transient_key_errors_only() { + let (retryable, completed) = split_delete_many_outcomes(vec![ + ObjectDeleteOutcome { + key: "slow".to_string(), + error: Some("SlowDown: reduce your request rate".to_string()), + }, + ObjectDeleteOutcome { + key: "internal".to_string(), + error: Some("InternalError: try again".to_string()), + }, + ObjectDeleteOutcome { + key: "denied".to_string(), + error: Some("AccessDenied: forbidden".to_string()), + }, + ObjectDeleteOutcome { + key: "ok".to_string(), + error: None, + }, + ]); + + assert_eq!( + retryable, + vec![ + ("slow".to_string(), "SlowDown: reduce your request rate".to_string()), + ("internal".to_string(), "InternalError: try again".to_string()) + ] + ); + assert_eq!( + completed, + vec![ + ObjectDeleteOutcome { + key: "denied".to_string(), + error: Some("AccessDenied: forbidden".to_string()) + }, + ObjectDeleteOutcome { + key: "ok".to_string(), + error: None + } + ] + ); + } + #[test] fn list_parts_xml_handles_array_single_part_and_pagination() { let xml = r#" diff --git a/packages/backend/native/src/runtime/storage_runtime/object_storage/error.rs b/packages/backend/native/src/runtime/storage_runtime/object_storage/error.rs index 0b2a35f21..8e5ddab29 100644 --- a/packages/backend/native/src/runtime/storage_runtime/object_storage/error.rs +++ b/packages/backend/native/src/runtime/storage_runtime/object_storage/error.rs @@ -51,6 +51,14 @@ impl ObjectStorageError { _ => false, } } + + pub(crate) fn is_retryable_http_status(&self) -> bool { + match self { + Self::Operation { source, .. } => source.is_retryable_http_status(), + Self::HttpStatus { status, .. } => *status == StatusCode::TOO_MANY_REQUESTS || status.is_server_error(), + _ => false, + } + } } pub(crate) type ObjectStorageResult = std::result::Result; diff --git a/packages/backend/native/src/runtime/storage_runtime/object_storage/types.rs b/packages/backend/native/src/runtime/storage_runtime/object_storage/types.rs index a6775a4f7..2e0baf367 100644 --- a/packages/backend/native/src/runtime/storage_runtime/object_storage/types.rs +++ b/packages/backend/native/src/runtime/storage_runtime/object_storage/types.rs @@ -36,6 +36,12 @@ pub(crate) struct ObjectListPage { pub(crate) next_continuation_token: Option, } +#[derive(Clone, Debug, PartialEq)] +pub(crate) struct ObjectDeleteOutcome { + pub(crate) key: String, + pub(crate) error: Option, +} + #[derive(Clone, Debug, PartialEq)] pub(crate) struct ObjectGetResult { pub(crate) body: Vec, diff --git a/packages/backend/server/src/core/storage/__tests__/blob-job.spec.ts b/packages/backend/server/src/core/storage/__tests__/blob-job.spec.ts index 842e00560..7dd1b778a 100644 --- a/packages/backend/server/src/core/storage/__tests__/blob-job.spec.ts +++ b/packages/backend/server/src/core/storage/__tests__/blob-job.spec.ts @@ -18,6 +18,7 @@ interface Context { add: Sinon.SinonStub; }; db: { + $queryRaw: Sinon.SinonStub; workspace: { findMany: Sinon.SinonStub; }; @@ -46,6 +47,7 @@ test.beforeEach(t => { add: Sinon.stub().resolves(undefined), }; t.context.db = { + $queryRaw: Sinon.stub(), workspace: { findMany: Sinon.stub(), }, @@ -81,6 +83,16 @@ const objectStorageRequiredCases: { context.event.emitAsync, ], }, + { + name: 'blob cleanup execution sweep', + run: context => context.job.executeBlobCleanupCandidatesByMarkedRuns({}), + untouched: context => [ + context.db.$queryRaw, + context.runtime.executeBlobCleanupCandidates, + context.event.emitAsync, + context.queue.add, + ], + }, { name: 'blob cleanup planning sweep', run: context => context.job.planUnreferencedWorkspaceBlobsBySid({}), @@ -213,3 +225,123 @@ test('blob cleanup planning drains each workspace cursor before continuing', asy ) ); }); + +test('daily blob cleanup execution uses a fixed job id', async t => { + await t.context.job.dailyBlobCleanupExecution(); + + t.true( + t.context.queue.add.calledWith( + 'backendRuntime.executeBlobCleanupCandidatesByMarkedRuns', + {}, + { jobId: 'daily-backend-runtime-blob-cleanup-execution' } + ) + ); +}); + +test('blob cleanup execution sweep drains marked runs and continues by page', async t => { + t.context.db.$queryRaw + .onFirstCall() + .resolves([{ runId: 'run-1' }, { runId: 'run-2' }]) + .onSecondCall() + .resolves([{ exists: true }]); + t.context.runtime.executeBlobCleanupCandidates + .onFirstCall() + .resolves({ + scannedCandidates: 100, + deletedObjects: 100, + deletedMetadata: 100, + skippedStillReferenced: 0, + failed: 0, + workspaceIds: ['workspace-1'], + }) + .onSecondCall() + .resolves({ + scannedCandidates: 5, + deletedObjects: 5, + deletedMetadata: 5, + skippedStillReferenced: 0, + failed: 0, + workspaceIds: ['workspace-1'], + }) + .onThirdCall() + .resolves({ + scannedCandidates: 1, + deletedObjects: 0, + deletedMetadata: 0, + skippedStillReferenced: 0, + failed: 1, + workspaceIds: [], + }); + + await t.context.job.executeBlobCleanupCandidatesByMarkedRuns({ + runLimit: 2, + gracePeriodDays: 14, + candidateLimit: 100, + }); + + t.is(t.context.runtime.executeBlobCleanupCandidates.callCount, 3); + t.deepEqual(t.context.runtime.executeBlobCleanupCandidates.firstCall.args, [ + 'run-1', + 14, + 100, + ]); + t.deepEqual(t.context.runtime.executeBlobCleanupCandidates.secondCall.args, [ + 'run-1', + 14, + 100, + ]); + t.deepEqual(t.context.runtime.executeBlobCleanupCandidates.thirdCall.args, [ + 'run-2', + 14, + 100, + ]); + t.is(t.context.event.emitAsync.callCount, 2); + t.true( + t.context.queue.add.calledWith( + 'backendRuntime.executeBlobCleanupCandidatesByMarkedRuns', + { runLimit: 2, gracePeriodDays: 14, candidateLimit: 100 } + ) + ); +}); + +test('blob cleanup execution sweep does not continue failed-only backlog', async t => { + t.context.db.$queryRaw + .onFirstCall() + .resolves([{ runId: 'run-1' }]) + .onSecondCall() + .resolves([{ exists: false }]); + t.context.runtime.executeBlobCleanupCandidates.resolves({ + scannedCandidates: 100, + deletedObjects: 0, + deletedMetadata: 0, + skippedStillReferenced: 0, + failed: 100, + workspaceIds: [], + }); + + await t.context.job.executeBlobCleanupCandidatesByMarkedRuns({ + runLimit: 1, + candidateLimit: 100, + }); + + t.is(t.context.runtime.executeBlobCleanupCandidates.callCount, 1); + t.false(t.context.queue.add.called); +}); + +test('blob cleanup execution sweep does not continue after drain errors', async t => { + t.context.db.$queryRaw + .onFirstCall() + .resolves([{ runId: 'run-1' }]) + .onSecondCall() + .resolves([{ exists: true }]); + t.context.runtime.executeBlobCleanupCandidates.rejects( + new Error('storage outage') + ); + + await t.context.job.executeBlobCleanupCandidatesByMarkedRuns({ + runLimit: 1, + }); + + t.is(t.context.runtime.executeBlobCleanupCandidates.callCount, 1); + t.false(t.context.queue.add.called); +}); diff --git a/packages/backend/server/src/core/storage/blob-job.ts b/packages/backend/server/src/core/storage/blob-job.ts index 2d25ce462..9008d482a 100644 --- a/packages/backend/server/src/core/storage/blob-job.ts +++ b/packages/backend/server/src/core/storage/blob-job.ts @@ -43,6 +43,11 @@ declare global { gracePeriodDays?: number; limit?: number; }; + 'backendRuntime.executeBlobCleanupCandidatesByMarkedRuns': { + runLimit?: number; + gracePeriodDays?: number; + candidateLimit?: number; + }; } } @@ -173,6 +178,21 @@ export class StorageBlobJob { }); } + async enqueueExecuteBlobCleanupCandidatesByMarkedRuns( + runLimit = 10, + gracePeriodDays = 30, + candidateLimit = 1000 + ) { + await this.queue.add( + 'backendRuntime.executeBlobCleanupCandidatesByMarkedRuns', + { + runLimit, + gracePeriodDays, + candidateLimit, + } + ); + } + @Cron(CronExpression.EVERY_DAY_AT_1AM) async dailyBlobMetadataBackfill() { await this.queue.add( @@ -200,6 +220,15 @@ export class StorageBlobJob { ); } + @Cron(CronExpression.EVERY_DAY_AT_4AM) + async dailyBlobCleanupExecution() { + await this.queue.add( + 'backendRuntime.executeBlobCleanupCandidatesByMarkedRuns', + {}, + { jobId: 'daily-backend-runtime-blob-cleanup-execution' } + ); + } + @OnJob('backendRuntime.backfillMissingBlobMetadata') async backfillMissingBlobMetadata({ workspaceId, @@ -357,6 +386,46 @@ export class StorageBlobJob { ); } + @OnJob('backendRuntime.executeBlobCleanupCandidatesByMarkedRuns') + async executeBlobCleanupCandidatesByMarkedRuns({ + runLimit = 10, + gracePeriodDays = 30, + candidateLimit = 1000, + }: Jobs['backendRuntime.executeBlobCleanupCandidatesByMarkedRuns']) { + if (!(await this.hasObjectStorage('blob cleanup execution sweep'))) { + return; + } + + const normalizedRunLimit = Math.max(1, runLimit); + const normalizedCandidateLimit = Math.max(1, candidateLimit); + const runIds = await this.loadPendingBlobCleanupRunIds(normalizedRunLimit); + let hadDrainError = false; + for (const runId of runIds) { + try { + await this.drainBlobCleanupExecution( + runId, + gracePeriodDays, + normalizedCandidateLimit + ); + } catch (err) { + hadDrainError = true; + this.logger.error(`blob cleanup execution failed run=${runId}`, err); + } + } + + if ( + !hadDrainError && + runIds.length === normalizedRunLimit && + (await this.hasMarkedBlobCleanupCandidates()) + ) { + await this.enqueueExecuteBlobCleanupCandidatesByMarkedRuns( + normalizedRunLimit, + gracePeriodDays, + normalizedCandidateLimit + ); + } + } + private async drainBlobMetadataBackfill( workspaceId: string, limit: number, @@ -421,6 +490,59 @@ export class StorageBlobJob { } } + private async drainBlobCleanupExecution( + runId: string, + gracePeriodDays: number, + limit: number + ) { + for (;;) { + const result = await this.rt.executeBlobCleanupCandidates( + runId, + gracePeriodDays, + limit + ); + await Promise.all( + result.workspaceIds.map((workspaceId: string) => + this.event.emitAsync('workspace.blobs.updated', { workspaceId }) + ) + ); + this.logger.log( + `executed blob cleanup run=${runId} deleted=${result.deletedObjects} skipped=${result.skippedStillReferenced} failed=${result.failed}` + ); + + const progressed = + result.deletedMetadata > 0 || result.skippedStillReferenced > 0; + if (result.scannedCandidates < limit || !progressed) { + break; + } + } + } + + private async loadPendingBlobCleanupRunIds(limit: number) { + const rows = await this.db.$queryRaw<{ runId: string }[]>` + SELECT run_id::text AS "runId" + FROM blob_cleanup_candidates + WHERE status IN ('marked', 'failed') + GROUP BY run_id + ORDER BY + CASE WHEN BOOL_OR(status = 'marked') THEN 0 ELSE 1 END ASC, + MIN(planned_at) ASC + LIMIT ${limit} + `; + return rows.map(row => row.runId); + } + + private async hasMarkedBlobCleanupCandidates() { + const rows = await this.db.$queryRaw<{ exists: boolean }[]>` + SELECT EXISTS( + SELECT 1 + FROM blob_cleanup_candidates + WHERE status = 'marked' + ) AS "exists" + `; + return rows[0]?.exists ?? false; + } + private async hasObjectStorage(operation: string) { const health = await this.rt.health(); if (health.provider) {