v0.2.13

src/db/blob_delete_outbox.rs

@@ -0,0 +1,155 @@
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use crate::db::pool::{Connection, params};
+
+#[derive(Debug, Clone)]
+pub struct BlobDeleteEvent {
+    pub id: String,
+    pub storage_key: String,
+    pub attempts: u32,
+}
+
+pub fn now_epoch() -> i64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_secs()
+        .min(i64::MAX as u64) as i64
+}
+
+pub fn enqueue(conn: &Connection, storage_key: &str) -> Result<BlobDeleteEvent, String> {
+    if storage_key.trim().is_empty() {
+        return Err("cannot enqueue an empty blob storage key".to_string());
+    }
+    let id = uuid::Uuid::new_v4().to_string();
+    let now = now_epoch();
+    let result = match crate::db::pool::configured_backend() {
+        crate::db::pool::DatabaseBackend::Mysql => conn.execute(
+            crate::dbq!(db_blob_delete_outbox::Q_001),
+            params![&id, storage_key, now, now],
+        ),
+        _ => conn.execute(
+            crate::dbq!(db_blob_delete_outbox::Q_001),
+            params![&id, storage_key, now],
+        ),
+    };
+    result.map_err(|e| format!("failed to enqueue blob deletion: {e}"))?;
+    Ok(BlobDeleteEvent {
+        id,
+        storage_key: storage_key.to_string(),
+        attempts: 0,
+    })
+}
+
+pub fn list_ready(conn: &Connection, limit: usize) -> Result<Vec<BlobDeleteEvent>, String> {
+    let now = now_epoch();
+    let mut stmt = conn
+        .prepare(crate::dbq!(db_blob_delete_outbox::Q_002))
+        .map_err(|e| format!("failed to prepare blob deletion outbox query: {e}"))?;
+    let query_params = match crate::db::pool::configured_backend() {
+        crate::db::pool::DatabaseBackend::Mysql => params![now, now, limit],
+        _ => params![now, limit],
+    };
+    let rows = stmt
+        .query_map(query_params, |row| {
+            Ok(BlobDeleteEvent {
+                id: row.get(0)?,
+                storage_key: row.get(1)?,
+                attempts: row.get::<_, i64>(2)?.max(0) as u32,
+            })
+        })
+        .map_err(|e| format!("failed to query blob deletion outbox: {e}"))?;
+    rows.map(|row| row.map_err(|e| format!("failed to read blob deletion outbox row: {e}")))
+        .collect()
+}
+
+pub fn claim(conn: &Connection, event: &BlobDeleteEvent, lease_secs: u64) -> Result<bool, String> {
+    let now = now_epoch();
+    let lease_until = now.saturating_add(lease_secs.max(1) as i64);
+    let changed = match crate::db::pool::configured_backend() {
+        crate::db::pool::DatabaseBackend::Mysql => conn.execute(
+            crate::dbq!(db_blob_delete_outbox::Q_003),
+            params![lease_until, &event.id, now, now],
+        ),
+        _ => conn.execute(
+            crate::dbq!(db_blob_delete_outbox::Q_003),
+            params![&event.id, lease_until, now],
+        ),
+    }
+    .map_err(|e| format!("failed to claim blob deletion: {e}"))?;
+    Ok(changed > 0)
+}
+
+pub fn mark_success(conn: &Connection, event: &BlobDeleteEvent) -> Result<bool, String> {
+    let changed = conn
+        .execute(
+            crate::dbq!(db_blob_delete_outbox::Q_004),
+            params![&event.id],
+        )
+        .map_err(|e| format!("failed to acknowledge blob deletion: {e}"))?;
+    Ok(changed > 0)
+}
+
+pub fn mark_failure(
+    conn: &Connection,
+    event: &BlobDeleteEvent,
+    error: &str,
+) -> Result<bool, String> {
+    let attempts = event.attempts.saturating_add(1);
+    let backoff = 2u64.saturating_pow(attempts.min(12)).min(3600);
+    let available_at = now_epoch().saturating_add(backoff as i64);
+    let now = now_epoch();
+    let truncated: String = error.chars().take(2048).collect();
+    let changed = match crate::db::pool::configured_backend() {
+        crate::db::pool::DatabaseBackend::Mysql => conn.execute(
+            crate::dbq!(db_blob_delete_outbox::Q_005),
+            params![attempts, available_at, &truncated, now, &event.id],
+        ),
+        _ => conn.execute(
+            crate::dbq!(db_blob_delete_outbox::Q_005),
+            params![&event.id, attempts, available_at, &truncated, now],
+        ),
+    }
+    .map_err(|e| format!("failed to update blob deletion retry state: {e}"))?;
+    Ok(changed > 0)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::db::pool::open_test_db;
+
+    #[test]
+    fn enqueue_participates_in_caller_transaction() {
+        let conn = open_test_db();
+        {
+            let transaction = conn.unchecked_transaction().unwrap();
+            enqueue(&transaction, "users/test/drafts/d/attachments/a").unwrap();
+        }
+        assert!(list_ready(&conn, 10).unwrap().is_empty());
+    }
+
+    #[test]
+    fn failed_deletions_are_retried_instead_of_dead_lettered() {
+        let conn = open_test_db();
+        let event = enqueue(&conn, "users/test/drafts/d/attachments/a").unwrap();
+        assert!(claim(&conn, &event, 60).unwrap());
+        assert!(mark_failure(&conn, &event, "storage unavailable").unwrap());
+        let count: i64 = conn
+            .query_row(
+                "SELECT COUNT(*) FROM blob_delete_outbox WHERE id=?1",
+                params![&event.id],
+                |row| row.get(0),
+            )
+            .unwrap();
+        assert_eq!(count, 1);
+    }
+
+    #[test]
+    fn successful_delete_is_acknowledged() {
+        let conn = open_test_db();
+        let event = enqueue(&conn, "users/test/drafts/d/attachments/a").unwrap();
+        assert!(mark_success(&conn, &event).unwrap());
+        assert!(list_ready(&conn, 10).unwrap().is_empty());
+    }
+}

src/db/drafts.rs

@@ -257,6 +257,67 @@ pub fn get_draft_attachments(
     Ok(attachments)
 }
 
+pub fn lock_draft_for_update(conn: &Connection, id: &str) -> Result<bool, String> {
+    match conn.query_row(crate::dbq!(db_drafts::Q_010), params![id], |_row| Ok(())) {
+        Ok(()) => Ok(true),
+        Err(crate::db::pool::DbError::QueryReturnedNoRows) => Ok(false),
+        Err(e) => Err(format!("Failed to lock draft: {e}")),
+    }
+}
+
+pub fn get_draft_attachment_for_update(
+    conn: &Connection,
+    draft_id: &str,
+    attachment_id: &str,
+) -> Result<Option<DraftAttachment>, String> {
+    let result = conn.query_row(
+        crate::dbq!(db_drafts::Q_011),
+        params![draft_id, attachment_id],
+        |row| {
+            Ok(DraftAttachment {
+                id: row.get(0)?,
+                draft_id: row.get(1)?,
+                filename: row.get(2)?,
+                content_type: row.get(3)?,
+                size: row.get(4)?,
+                storage_key: row.get(5)?,
+                checksum_sha256: row.get(6)?,
+                created_at: row.get(7)?,
+            })
+        },
+    );
+    match result {
+        Ok(attachment) => Ok(Some(attachment)),
+        Err(crate::db::pool::DbError::QueryReturnedNoRows) => Ok(None),
+        Err(e) => Err(format!("Failed to lock draft attachment: {e}")),
+    }
+}
+
+pub fn get_draft_attachments_for_update(
+    conn: &Connection,
+    draft_id: &str,
+) -> Result<Vec<DraftAttachment>, String> {
+    let mut stmt = conn
+        .prepare(crate::dbq!(db_drafts::Q_012))
+        .map_err(|e| format!("Failed to prepare locked draft attachments query: {e}"))?;
+    let rows = stmt
+        .query_map(params![draft_id], |row| {
+            Ok(DraftAttachment {
+                id: row.get(0)?,
+                draft_id: row.get(1)?,
+                filename: row.get(2)?,
+                content_type: row.get(3)?,
+                size: row.get(4)?,
+                storage_key: row.get(5)?,
+                checksum_sha256: row.get(6)?,
+                created_at: row.get(7)?,
+            })
+        })
+        .map_err(|e| format!("Failed to query locked draft attachments: {e}"))?;
+    rows.map(|row| row.map_err(|e| format!("Failed to read locked attachment row: {e}")))
+        .collect()
+}
+
 pub fn delete_draft_attachment(conn: &Connection, id: &str) -> Result<bool, String> {
     let deleted = conn
         .execute(crate::dbq!(db_drafts::Q_008), params![id])

src/db/mod.rs

@@ -1,3 +1,4 @@
+pub mod blob_delete_outbox;
 pub mod calendar;
 pub mod contact_groups;
 pub mod contacts;

src/db/pool.rs

@@ -30,6 +30,10 @@ const SQLITE_MIGRATIONS: &[(u32, &str)] = &[
         48,
         include_str!("../../migrations/sqlite/V048__folder_counter_generation.sql"),
     ),
+    (
+        49,
+        include_str!("../../migrations/sqlite/V049__blob_delete_outbox.sql"),
+    ),
 ];
 const MYSQL_MIGRATIONS: &[(u32, &str)] = &[
     (
@@ -40,6 +44,10 @@ const MYSQL_MIGRATIONS: &[(u32, &str)] = &[
         48,
         include_str!("../../migrations/mysql/V048__folder_counter_generation.sql"),
     ),
+    (
+        49,
+        include_str!("../../migrations/mysql/V049__blob_delete_outbox.sql"),
+    ),
 ];
 const PGSQL_MIGRATIONS: &[(u32, &str)] = &[
     (
@@ -50,9 +58,14 @@ const PGSQL_MIGRATIONS: &[(u32, &str)] = &[
         48,
         include_str!("../../migrations/pgsql/V048__folder_counter_generation.sql"),
     ),
+    (
+        49,
+        include_str!("../../migrations/pgsql/V049__blob_delete_outbox.sql"),
+    ),
 ];
 
 const TENANT_TABLES: &[&str] = &[
+    "blob_delete_outbox",
     "calendar_events",
     "calendar_settings",
     "contact_group_members",
@@ -83,6 +96,7 @@ const TENANT_TABLES: &[&str] = &[
 ];
 
 const IMPORT_TABLES: &[&str] = &[
+    "blob_delete_outbox",
     "calendar_events",
     "calendar_settings",
     "contact_groups",

src/main.rs

@@ -121,6 +121,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             .map_err(std::io::Error::other)?;
     }
     let _search_worker = search_engine.start_worker();
+    let _blob_delete_worker =
+        crate::storage::start_blob_delete_worker(config.clone(), storage.clone());
 
     let event_bus = Arc::new(realtime::events::EventBus::new());
     let idle_manager = Arc::new(realtime::idle::IdleManager::new());

src/queries/mysql/queries.rs

@@ -167,6 +167,9 @@ pub mod db_drafts {
     pub const Q_008: &str = "DELETE FROM draft_attachments WHERE id = ?";
     pub const Q_009: &str =
         "UPDATE draft_attachments SET file_path=?, checksum_sha256=? WHERE id=?";
+    pub const Q_010: &str = "SELECT id FROM drafts WHERE id=? FOR UPDATE";
+    pub const Q_011: &str = "SELECT id, draft_id, filename, content_type, size, file_path, checksum_sha256, created_at FROM draft_attachments WHERE draft_id=? AND id=? FOR UPDATE";
+    pub const Q_012: &str = "SELECT id, draft_id, filename, content_type, size, file_path, checksum_sha256, created_at FROM draft_attachments WHERE draft_id=? ORDER BY created_at ASC FOR UPDATE";
 }
 
 pub mod db_folder_preferences {
@@ -527,3 +530,11 @@ pub mod db_search_outbox {
     pub const Q_006: &str = "UPDATE search_outbox SET claimed_until=? WHERE id=? AND generation=? AND dead_letter=0 AND available_at <= ? AND claimed_until <= ?";
     pub const Q_007: &str = "DELETE FROM search_outbox WHERE updated_at < ?";
 }
+
+pub mod db_blob_delete_outbox {
+    pub const Q_001: &str = "INSERT INTO blob_delete_outbox (id, storage_key, attempts, available_at, claimed_until, last_error, created_at, updated_at) VALUES (?, ?, 0, 0, 0, NULL, ?, ?)";
+    pub const Q_002: &str = "SELECT id, storage_key, attempts FROM blob_delete_outbox WHERE available_at <= ? AND claimed_until <= ? ORDER BY updated_at ASC LIMIT ?";
+    pub const Q_003: &str = "UPDATE blob_delete_outbox SET claimed_until=? WHERE id=? AND available_at <= ? AND claimed_until <= ?";
+    pub const Q_004: &str = "DELETE FROM blob_delete_outbox WHERE id=?";
+    pub const Q_005: &str = "UPDATE blob_delete_outbox SET attempts=?, available_at=?, claimed_until=0, last_error=?, updated_at=? WHERE id=?";
+}

src/queries/pgsql/queries.rs

@@ -168,6 +168,9 @@ pub mod db_drafts {
     pub const Q_008: &str = "DELETE FROM draft_attachments WHERE id = $1";
     pub const Q_009: &str =
         "UPDATE draft_attachments SET file_path=$1, checksum_sha256=$2 WHERE id=$3";
+    pub const Q_010: &str = "SELECT id FROM drafts WHERE id=$1 FOR UPDATE";
+    pub const Q_011: &str = "SELECT id, draft_id, filename, content_type, size, file_path, checksum_sha256, created_at FROM draft_attachments WHERE draft_id=$1 AND id=$2 FOR UPDATE";
+    pub const Q_012: &str = "SELECT id, draft_id, filename, content_type, size, file_path, checksum_sha256, created_at FROM draft_attachments WHERE draft_id=$1 ORDER BY created_at ASC FOR UPDATE";
 }
 
 pub mod db_folder_preferences {
@@ -528,3 +531,11 @@ pub mod db_search_outbox {
     pub const Q_006: &str = "UPDATE search_outbox SET claimed_until=$3 WHERE id=$1 AND generation=$2 AND dead_letter=0 AND available_at <= $4 AND claimed_until <= $4";
     pub const Q_007: &str = "DELETE FROM search_outbox WHERE updated_at < $1";
 }
+
+pub mod db_blob_delete_outbox {
+    pub const Q_001: &str = "INSERT INTO blob_delete_outbox (id, storage_key, attempts, available_at, claimed_until, last_error, created_at, updated_at) VALUES ($1, $2, 0, 0, 0, NULL, $3, $3)";
+    pub const Q_002: &str = "SELECT id, storage_key, attempts FROM blob_delete_outbox WHERE available_at <= $1 AND claimed_until <= $1 ORDER BY updated_at ASC LIMIT $2";
+    pub const Q_003: &str = "UPDATE blob_delete_outbox SET claimed_until=$2 WHERE id=$1 AND available_at <= $3 AND claimed_until <= $3";
+    pub const Q_004: &str = "DELETE FROM blob_delete_outbox WHERE id=$1";
+    pub const Q_005: &str = "UPDATE blob_delete_outbox SET attempts=$2, available_at=$3, claimed_until=0, last_error=$4, updated_at=$5 WHERE id=$1";
+}

src/queries/sqlite/queries.rs

@@ -165,6 +165,9 @@ pub mod db_drafts {
     pub const Q_008: &str = "DELETE FROM draft_attachments WHERE id = ?1";
     pub const Q_009: &str =
         "UPDATE draft_attachments SET file_path=?1, checksum_sha256=?2 WHERE id=?3";
+    pub const Q_010: &str = "SELECT id FROM drafts WHERE id=?1";
+    pub const Q_011: &str = "SELECT id, draft_id, filename, content_type, size, file_path, checksum_sha256, created_at FROM draft_attachments WHERE draft_id=?1 AND id=?2";
+    pub const Q_012: &str = "SELECT id, draft_id, filename, content_type, size, file_path, checksum_sha256, created_at FROM draft_attachments WHERE draft_id=?1 ORDER BY created_at ASC";
 }
 
 pub mod db_folder_preferences {
@@ -782,3 +785,11 @@ pub mod db_search_outbox {
     pub const Q_006: &str = "UPDATE search_outbox SET claimed_until=?3 WHERE id=?1 AND generation=?2 AND dead_letter=0 AND available_at <= ?4 AND claimed_until <= ?4";
     pub const Q_007: &str = "DELETE FROM search_outbox WHERE updated_at < ?1";
 }
+
+pub mod db_blob_delete_outbox {
+    pub const Q_001: &str = "INSERT INTO blob_delete_outbox (id, storage_key, attempts, available_at, claimed_until, last_error, created_at, updated_at) VALUES (?1, ?2, 0, 0, 0, NULL, ?3, ?3)";
+    pub const Q_002: &str = "SELECT id, storage_key, attempts FROM blob_delete_outbox WHERE available_at <= ?1 AND claimed_until <= ?1 ORDER BY updated_at ASC LIMIT ?2";
+    pub const Q_003: &str = "UPDATE blob_delete_outbox SET claimed_until=?2 WHERE id=?1 AND available_at <= ?3 AND claimed_until <= ?3";
+    pub const Q_004: &str = "DELETE FROM blob_delete_outbox WHERE id=?1";
+    pub const Q_005: &str = "UPDATE blob_delete_outbox SET attempts=?2, available_at=?3, claimed_until=0, last_error=?4, updated_at=?5 WHERE id=?1";
+}

src/routes/attachments.rs

@@ -11,7 +11,7 @@ use crate::auth::session::SessionState;
 use crate::config::AppConfig;
 use crate::db;
 use crate::error::AppError;
-use crate::storage::{BlobStorage, attachment_key};
+use crate::storage::{self, BlobStorage, attachment_key};
 
 #[derive(Debug, Serialize)]
 struct UploadResponse {
@@ -73,7 +73,13 @@ pub async fn upload_attachment(
             let chunk = match field.chunk().await {
                 Ok(chunk) => chunk,
                 Err(e) => {
-                    let _ = writer.abort().await;
+                    if writer.abort().await.is_err()
+                        && let Err(cleanup_error) =
+                            storage::enqueue_unreferenced_delete(&conn, storage.as_ref(), &key)
+                                .await
+                    {
+                        tracing::error!(error = %cleanup_error, storage_key = %key, "Failed to queue aborted attachment cleanup");
+                    }
                     return Err(AppError::BadRequest(format!(
                         "Failed to read field data: {e}"
                     )));
@@ -82,7 +88,12 @@ pub async fn upload_attachment(
             let Some(chunk) = chunk else { break };
             size = size.saturating_add(chunk.len());
             if size > max_attachment_size {
-                let _ = writer.abort().await;
+                if writer.abort().await.is_err()
+                    && let Err(cleanup_error) =
+                        storage::enqueue_unreferenced_delete(&conn, storage.as_ref(), &key).await
+                {
+                    tracing::error!(error = %cleanup_error, storage_key = %key, "Failed to queue oversized attachment cleanup");
+                }
                 return Err(AppError::BadRequest(format!(
                     "File '{}' exceeds maximum size of {} MiB",
                     filename, config.max_attachment_size_mb
@@ -90,18 +101,37 @@ pub async fn upload_attachment(
             }
             checksum.update(&chunk);
             if let Err(e) = writer.write(chunk).await {
-                let _ = writer.abort().await;
+                if writer.abort().await.is_err()
+                    && let Err(cleanup_error) =
+                        storage::enqueue_unreferenced_delete(&conn, storage.as_ref(), &key).await
+                {
+                    tracing::error!(error = %cleanup_error, storage_key = %key, "Failed to queue interrupted attachment cleanup");
+                }
                 return Err(AppError::ServiceUnavailable(format!(
                     "Failed to upload attachment: {e}"
                 )));
             }
         }
 
-        let stat = writer.finish().await.map_err(|e| {
-            AppError::ServiceUnavailable(format!("Failed to finalize attachment upload: {e}"))
-        })?;
+        let stat = match writer.finish().await {
+            Ok(stat) => stat,
+            Err(e) => {
+                if let Err(cleanup_error) =
+                    storage::enqueue_unreferenced_delete(&conn, storage.as_ref(), &key).await
+                {
+                    tracing::error!(error = %cleanup_error, storage_key = %key, "Failed to queue failed-finalize attachment cleanup");
+                }
+                return Err(AppError::ServiceUnavailable(format!(
+                    "Failed to finalize attachment upload: {e}"
+                )));
+            }
+        };
         if stat.size != size as u64 {
-            let _ = storage.delete(&key).await;
+            if let Err(cleanup_error) =
+                storage::enqueue_unreferenced_delete(&conn, storage.as_ref(), &key).await
+            {
+                tracing::error!(error = %cleanup_error, storage_key = %key, "Failed to queue invalid attachment cleanup");
+            }
             return Err(AppError::InternalError(format!(
                 "Attachment size verification failed for '{filename}'"
             )));
@@ -118,7 +148,11 @@ pub async fn upload_attachment(
             checksum_sha256: Some(&checksum),
         };
         if let Err(e) = db::drafts::add_draft_attachment(&conn, &attachment) {
-            let _ = storage.delete(&key).await;
+            if let Err(cleanup_error) =
+                storage::enqueue_unreferenced_delete(&conn, storage.as_ref(), &key).await
+            {
+                tracing::error!(error = %cleanup_error, storage_key = %key, "Failed to queue unattached blob cleanup");
+            }
             return Err(AppError::InternalError(e));
         }
 
@@ -180,21 +214,30 @@ pub async fn delete_attachment(
     let conn = db::pool::open_user_db(&config.data_dir, &session.user_hash)
         .map_err(|e| AppError::InternalError(format!("Failed to open database: {e}")))?;
 
-    let attachments =
-        db::drafts::get_draft_attachments(&conn, &draft_id).map_err(AppError::InternalError)?;
-    let attachment = attachments
-        .iter()
-        .find(|a| a.id == attachment_id)
-        .ok_or_else(|| AppError::NotFound("Attachment not found".to_string()))?;
-
+    let transaction = conn.unchecked_transaction().map_err(|e| {
+        AppError::InternalError(format!(
+            "Failed to start attachment delete transaction: {e}"
+        ))
+    })?;
+    let attachment =
+        db::drafts::get_draft_attachment_for_update(&transaction, &draft_id, &attachment_id)
+            .map_err(AppError::InternalError)?
+            .ok_or_else(|| AppError::NotFound("Attachment not found".to_string()))?;
     let storage_key = attachment.storage_key.clone();
-    let deleted = db::drafts::delete_draft_attachment(&conn, &attachment_id)
+    let cleanup_event = db::blob_delete_outbox::enqueue(&transaction, &storage_key)
+        .map_err(AppError::InternalError)?;
+    let deleted = db::drafts::delete_draft_attachment(&transaction, &attachment_id)
         .map_err(AppError::InternalError)?;
     if !deleted {
         return Err(AppError::NotFound("Attachment not found".to_string()));
     }
-    if let Err(error) = storage.delete(&storage_key).await {
-        tracing::warn!(error = %error, storage_key = %storage_key, "Failed to remove deleted attachment blob");
+    transaction.commit().map_err(|e| {
+        AppError::InternalError(format!("Failed to commit attachment deletion: {e}"))
+    })?;
+    if let Err(error) =
+        storage::process_queued_delete(&conn, storage.as_ref(), &cleanup_event).await
+    {
+        tracing::warn!(error = %error, storage_key = %storage_key, "Failed to process queued attachment blob deletion");
     }
 
     Ok(Json(DeleteResponse {

src/routes/drafts.rs

@@ -1179,25 +1179,44 @@ pub async fn delete_draft_handler(
             delete_imap_draft_source(imap_client.as_ref(), &creds, &drafts_folder, source).await?;
     }
 
-    let attachments = if existing.is_some() {
-        db::drafts::get_draft_attachments(&conn, &id).map_err(AppError::InternalError)?
-    } else {
-        Vec::new()
-    };
-    if existing.is_some() {
-        let deleted = db::drafts::delete_draft(&conn, &id).map_err(AppError::InternalError)?;
+    let (attachments, cleanup_events) = if existing.is_some() {
+        let transaction = conn.unchecked_transaction().map_err(|e| {
+            AppError::InternalError(format!("Failed to start draft delete transaction: {e}"))
+        })?;
+        if !db::drafts::lock_draft_for_update(&transaction, &id).map_err(AppError::InternalError)? {
+            return Err(AppError::NotFound("Draft not found".to_string()));
+        }
+        let attachments = db::drafts::get_draft_attachments_for_update(&transaction, &id)
+            .map_err(AppError::InternalError)?;
+        let mut events = Vec::with_capacity(attachments.len());
+        for attachment in &attachments {
+            events.push(
+                db::blob_delete_outbox::enqueue(&transaction, &attachment.storage_key)
+                    .map_err(AppError::InternalError)?,
+            );
+        }
+        let deleted =
+            db::drafts::delete_draft(&transaction, &id).map_err(AppError::InternalError)?;
         if !deleted {
             return Err(AppError::NotFound("Draft not found".to_string()));
         }
-    }
+        transaction.commit().map_err(|e| {
+            AppError::InternalError(format!("Failed to commit draft deletion: {e}"))
+        })?;
+        (attachments, events)
+    } else {
+        (Vec::new(), Vec::new())
+    };
+
     let mut attachments_deleted = 0usize;
-    for attachment in &attachments {
-        match storage.delete(&attachment.storage_key).await {
-            Ok(()) => attachments_deleted += 1,
+    for (attachment, event) in attachments.iter().zip(cleanup_events.iter()) {
+        match storage::process_queued_delete(&conn, storage.as_ref(), event).await {
+            Ok(true) => attachments_deleted += 1,
+            Ok(false) => {}
             Err(error) => tracing::warn!(
                 error = %error,
                 storage_key = %attachment.storage_key,
-                "Failed to remove deleted draft attachment blob"
+                "Failed to process queued draft attachment deletion"
             ),
         }
     }

src/routes/send.rs

@@ -457,26 +457,65 @@ async fn load_draft_attachments(
 }
 
 async fn cleanup_draft(data_dir: &str, user_hash: &str, draft_id: &str, storage: &dyn BlobStorage) {
-    let Ok(conn) = db::pool::open_user_db(data_dir, user_hash) else {
-        return;
+    let conn = match db::pool::open_user_db(data_dir, user_hash) {
+        Ok(conn) => conn,
+        Err(error) => {
+            tracing::warn!(error = %error, "Failed to open database for draft cleanup after send");
+            return;
+        }
     };
-    let attachments = db::drafts::get_draft_attachments(&conn, draft_id).unwrap_or_default();
-    let deleted = match db::drafts::delete_draft(&conn, draft_id) {
+    let transaction = match conn.unchecked_transaction() {
+        Ok(transaction) => transaction,
+        Err(error) => {
+            tracing::warn!(error = %error, "Failed to start draft cleanup transaction after send");
+            return;
+        }
+    };
+    match db::drafts::lock_draft_for_update(&transaction, draft_id) {
+        Ok(true) => {}
+        Ok(false) => return,
+        Err(error) => {
+            tracing::warn!(error = %error, "Failed to lock draft for cleanup after send");
+            return;
+        }
+    }
+    let attachments = match db::drafts::get_draft_attachments_for_update(&transaction, draft_id) {
+        Ok(attachments) => attachments,
+        Err(error) => {
+            tracing::warn!(error = %error, "Failed to load draft attachments for cleanup after send");
+            return;
+        }
+    };
+    let mut cleanup_events = Vec::with_capacity(attachments.len());
+    for attachment in &attachments {
+        match db::blob_delete_outbox::enqueue(&transaction, &attachment.storage_key) {
+            Ok(event) => cleanup_events.push(event),
+            Err(error) => {
+                tracing::warn!(error = %error, storage_key = %attachment.storage_key, "Failed to queue draft attachment cleanup after send");
+                return;
+            }
+        }
+    }
+    let deleted = match db::drafts::delete_draft(&transaction, draft_id) {
         Ok(deleted) => deleted,
-        Err(e) => {
-            tracing::warn!(error = %e, "Failed to delete draft after send");
+        Err(error) => {
+            tracing::warn!(error = %error, "Failed to delete draft after send");
             return;
         }
     };
     if !deleted {
         return;
     }
-    for attachment in attachments {
-        if let Err(e) = storage.delete(&attachment.storage_key).await {
+    if let Err(error) = transaction.commit() {
+        tracing::warn!(error = %error, "Failed to commit draft cleanup after send");
+        return;
+    }
+    for (attachment, event) in attachments.iter().zip(cleanup_events.iter()) {
+        if let Err(error) = storage::process_queued_delete(&conn, storage, event).await {
             tracing::warn!(
-                error = %e,
+                error = %error,
                 storage_key = %attachment.storage_key,
-                "Failed to delete draft attachment after send"
+                "Failed to process queued draft attachment cleanup after send"
             );
         }
     }

src/storage/mod.rs

@@ -160,13 +160,17 @@ impl BlobStorage for ObjectStorage {
     }
 
     async fn health(&self) -> Result<(), String> {
-        let key = format!("_health/{}", uuid::Uuid::new_v4());
+        let key = "_health/probe".to_string();
         let mut upload = self.begin_upload(&key).await?;
         upload.write(Bytes::from_static(b"ok")).await?;
         let stat = upload.finish().await?;
         if stat.size != 2 {
-            let _ = self.delete(&key).await;
-            return Err("storage health probe returned unexpected size".to_string());
+            return match self.delete(&key).await {
+                Ok(()) => Err("storage health probe returned unexpected size".to_string()),
+                Err(cleanup_error) => Err(format!(
+                    "storage health probe returned unexpected size and cleanup failed: {cleanup_error}"
+                )),
+            };
         }
         self.delete(&key).await?;
         Ok(())
@@ -311,6 +315,89 @@ pub fn from_config(config: &AppConfig) -> Result<Arc<dyn BlobStorage>, String> {
     }
 }
 
+const BLOB_DELETE_BATCH_SIZE: usize = 100;
+const BLOB_DELETE_LEASE_SECS: u64 = 60;
+const BLOB_DELETE_WORKER_INTERVAL_SECS: u64 = 30;
+
+pub async fn process_queued_delete(
+    conn: &crate::db::pool::Connection,
+    storage: &dyn BlobStorage,
+    event: &crate::db::blob_delete_outbox::BlobDeleteEvent,
+) -> Result<bool, String> {
+    if !crate::db::blob_delete_outbox::claim(conn, event, BLOB_DELETE_LEASE_SECS)? {
+        return Ok(false);
+    }
+    match storage.delete(&event.storage_key).await {
+        Ok(()) => {
+            crate::db::blob_delete_outbox::mark_success(conn, event)?;
+            Ok(true)
+        }
+        Err(error) => {
+            crate::db::blob_delete_outbox::mark_failure(conn, event, &error)?;
+            tracing::warn!(
+                error = %error,
+                storage_key = %event.storage_key,
+                attempts = event.attempts.saturating_add(1),
+                "Blob deletion failed; cleanup remains queued"
+            );
+            Ok(false)
+        }
+    }
+}
+
+pub async fn enqueue_unreferenced_delete(
+    conn: &crate::db::pool::Connection,
+    storage: &dyn BlobStorage,
+    storage_key: &str,
+) -> Result<bool, String> {
+    match crate::db::blob_delete_outbox::enqueue(conn, storage_key) {
+        Ok(event) => process_queued_delete(conn, storage, &event).await,
+        Err(queue_error) => match storage.delete(storage_key).await {
+            Ok(()) => Ok(true),
+            Err(delete_error) => Err(format!(
+                "failed to persist blob cleanup ({queue_error}); direct delete also failed: {delete_error}"
+            )),
+        },
+    }
+}
+
+pub async fn process_blob_delete_outbox(
+    config: &AppConfig,
+    storage: &dyn BlobStorage,
+) -> Result<usize, String> {
+    let mut processed = 0usize;
+    for user_hash in crate::db::pool::list_account_user_hashes()? {
+        let conn = crate::db::pool::open_user_db(&config.data_dir, &user_hash)?;
+        let events = crate::db::blob_delete_outbox::list_ready(&conn, BLOB_DELETE_BATCH_SIZE)?;
+        for event in events {
+            if process_queued_delete(&conn, storage, &event).await? {
+                processed = processed.saturating_add(1);
+            }
+        }
+    }
+    Ok(processed)
+}
+
+pub fn start_blob_delete_worker(
+    config: Arc<AppConfig>,
+    storage: Arc<dyn BlobStorage>,
+) -> tokio::task::JoinHandle<()> {
+    tokio::spawn(async move {
+        loop {
+            match process_blob_delete_outbox(&config, storage.as_ref()).await {
+                Ok(processed) if processed > 0 => {
+                    tracing::debug!(processed, "Processed queued blob deletions");
+                }
+                Ok(_) => {}
+                Err(error) => {
+                    tracing::warn!(%error, "Blob deletion worker pass failed");
+                }
+            }
+            tokio::time::sleep(Duration::from_secs(BLOB_DELETE_WORKER_INTERVAL_SECS)).await;
+        }
+    })
+}
+
 pub async fn read_all(storage: &dyn BlobStorage, key: &str) -> Result<Vec<u8>, String> {
     let mut stream = storage.get(key).await?;
     let mut data = Vec::new();
@@ -537,7 +624,11 @@ pub async fn migrate_attachments(
                 let source_checksum = format!("{:x}", hasher.finalize());
                 let target_stat = upload.finish().await?;
                 if target_stat.size != source_meta.len() {
-                    let _ = destination.delete(&target_key).await;
+                    if let Err(cleanup_error) =
+                        enqueue_unreferenced_delete(&conn, destination.as_ref(), &target_key).await
+                    {
+                        eprintln!("fail {target_key}: cleanup queue failed: {cleanup_error}");
+                    }
                     report.failed += 1;
                     eprintln!("fail {target_key}: size verification failed");
                     continue;
@@ -548,14 +639,29 @@ pub async fn migrate_attachments(
                     {
                         Ok(checksum) => checksum,
                         Err(error) => {
-                            let _ = destination.delete(&target_key).await;
+                            if let Err(cleanup_error) = enqueue_unreferenced_delete(
+                                &conn,
+                                destination.as_ref(),
+                                &target_key,
+                            )
+                            .await
+                            {
+                                eprintln!(
+                                    "fail {target_key}: cleanup queue failed: {cleanup_error}"
+                                );
+                            }
                             report.failed += 1;
                             eprintln!("fail {target_key}: checksum verification failed: {error}");
                             continue;
                         }
                     };
                     if target_checksum != source_checksum {
-                        let _ = destination.delete(&target_key).await;
+                        if let Err(cleanup_error) =
+                            enqueue_unreferenced_delete(&conn, destination.as_ref(), &target_key)
+                                .await
+                        {
+                            eprintln!("fail {target_key}: cleanup queue failed: {cleanup_error}");
+                        }
                         report.failed += 1;
                         eprintln!("fail {target_key}: checksum verification failed");
                         continue;
@@ -567,7 +673,11 @@ pub async fn migrate_attachments(
                     &target_key,
                     Some(&source_checksum),
                 ) {
-                    let _ = destination.delete(&target_key).await;
+                    if let Err(cleanup_error) =
+                        enqueue_unreferenced_delete(&conn, destination.as_ref(), &target_key).await
+                    {
+                        tracing::error!(error = %cleanup_error, storage_key = %target_key, "Failed to queue storage migration cleanup");
+                    }
                     return Err(error);
                 }
                 report.migrated += 1;