feat(server): batch blob gc (#15183)

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 }

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<String>, 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::<Vec<_>>();
+      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::<HashMap<_, _>>();
+
+      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)
   }

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<T> = RuntimeResult<T>;
 
@@ -680,6 +684,15 @@ impl StorageRuntime {
     }
   }
 
+  pub(crate) async fn object_storage_delete_many(&self, keys: Vec<String>) -> Result<Vec<ObjectDeleteOutcome>> {
+    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<String>) -> Vec<ObjectDeleteOutcome> {
+  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<String>) -> Result<Vec<ObjectDeleteOutcome>> {
+  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<String>) -> Result<Vec<ObjectDeleteOutcome>> {
+  let client = config.build_client()?;
+  let mut chunks = keys
+    .chunks(OBJECT_DELETE_MANY_CHUNK_SIZE)
+    .map(|chunk| chunk.to_vec())
+    .collect::<Vec<_>>()
+    .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<Option<ObjectMetadata>> {
   let raw = match fs::read_to_string(PathBuf::from(format!("{}.metadata.json", path.display()))) {
     Ok(raw) => raw,

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<String>) -> ObjectStorageResult<Vec<ObjectDeleteOutcome>> {
+    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<Vec<ObjectDeleteOutcome>> {
+    let objects = keys
+      .iter()
+      .map(|key| ObjectIdentifier::new(key.clone()))
+      .collect::<Vec<_>>();
+    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::<Vec<_>>();
+    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<String, String>) {
@@ -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<ObjectDeleteOutcome>) -> (Vec<(String, String)>, Vec<ObjectDeleteOutcome>) {
+  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('&', "&amp;").replace('<', "&lt;").replace('>', "&gt;")
 }
@@ -734,6 +864,56 @@ mod tests {
     assert!(!is_not_found_body(b"<Error><Code>AccessDenied</Code></Error>"));
   }
 
+  #[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#"<?xml version="1.0" encoding="UTF-8"?>

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<T> = std::result::Result<T, ObjectStorageError>;

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<String>,
 }
 
+#[derive(Clone, Debug, PartialEq)]
+pub(crate) struct ObjectDeleteOutcome {
+  pub(crate) key: String,
+  pub(crate) error: Option<String>,
+}
+
 #[derive(Clone, Debug, PartialEq)]
 pub(crate) struct ObjectGetResult {
   pub(crate) body: Vec<u8>,

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);
+});

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) {