fix: fence worker retries and object publication

migrations/015_worker_attempt_leases.sql

@@ -0,0 +1,17 @@
+ALTER TABLE tasks
+  ADD COLUMN IF NOT EXISTS processing_attempt BIGINT NOT NULL DEFAULT 0,
+  ADD COLUMN IF NOT EXISTS lease_owner UUID,
+  ADD COLUMN IF NOT EXISTS lease_until TIMESTAMPTZ;
+
+ALTER TABLE task_files
+  ADD COLUMN IF NOT EXISTS processing_attempt BIGINT NOT NULL DEFAULT 0,
+  ADD COLUMN IF NOT EXISTS lease_owner UUID,
+  ADD COLUMN IF NOT EXISTS lease_until TIMESTAMPTZ;
+
+CREATE INDEX IF NOT EXISTS idx_tasks_processing_lease
+  ON tasks(lease_until)
+  WHERE status = 'processing';
+
+CREATE INDEX IF NOT EXISTS idx_task_files_processing_lease
+  ON task_files(task_id, lease_until)
+  WHERE status = 'processing';

src/api/admin.rs

@@ -556,28 +556,14 @@ async fn cancel_task(
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "开启事务失败").with_source(err))?;
 
-    sqlx::query(
-        r#"
-        UPDATE task_files
-        SET status = 'failed',
-            error_message = '任务已取消',
-            completed_at = NOW()
-        WHERE task_id = $1 AND status IN ('pending', 'processing')
-        "#,
-    )
-    .bind(task_id)
-    .execute(&mut *tx)
-    .await
-    .map_err(|err| AppError::new(ErrorCode::Internal, "更新任务文件失败").with_source(err))?;
-
     let updated = sqlx::query(
         r#"
         UPDATE tasks
         SET status = 'cancelled',
             error_message = '管理员取消任务',
             completed_at = NOW(),
-            completed_files = (SELECT COUNT(*) FROM task_files WHERE task_id = $1 AND status = 'completed'),
+            lease_owner = NULL,
-            failed_files = (SELECT COUNT(*) FROM task_files WHERE task_id = $1 AND status = 'failed')
+            lease_until = NULL
         WHERE id = $1 AND status IN ('pending', 'processing')
         "#,
     )
@@ -590,6 +576,44 @@ async fn cancel_task(
         return Err(AppError::new(ErrorCode::InvalidRequest, "任务状态无法取消"));
     }
 
+    sqlx::query(
+        r#"
+        UPDATE task_files
+        SET status = 'failed',
+            error_message = '任务已取消',
+            completed_at = NOW(),
+            lease_owner = NULL,
+            lease_until = NULL
+        WHERE task_id = $1 AND status IN ('pending', 'processing')
+        "#,
+    )
+    .bind(task_id)
+    .execute(&mut *tx)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "更新任务文件失败").with_source(err))?;
+
+    sqlx::query(
+        r#"
+        UPDATE tasks
+        SET completed_files = (
+              SELECT COUNT(*) FROM task_files WHERE task_id = $1 AND status = 'completed'
+            ),
+            failed_files = (
+              SELECT COUNT(*) FROM task_files WHERE task_id = $1 AND status = 'failed'
+            ),
+            total_compressed_size = COALESCE((
+              SELECT SUM(compressed_size)
+              FROM task_files
+              WHERE task_id = $1 AND status = 'completed'
+            ), 0)::bigint
+        WHERE id = $1 AND status = 'cancelled'
+        "#,
+    )
+    .bind(task_id)
+    .execute(&mut *tx)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "更新任务统计失败").with_source(err))?;
+
     tx.commit()
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "提交事务失败").with_source(err))?;

src/services/storage.rs

@@ -291,6 +291,26 @@ pub fn result_key(retention_hours: i64, task_id: Uuid, file_id: Uuid, extension:
     )
 }
 
+pub fn result_attempt_key(
+    retention_hours: i64,
+    task_id: Uuid,
+    file_id: Uuid,
+    task_attempt: i64,
+    file_attempt: i64,
+    extension: &str,
+) -> String {
+    let now = Utc::now();
+    format!(
+        "results/{}/{:04}/{:02}/{task_id}/{file_id}-t{}-f{}.{}",
+        retention_prefix(retention_hours),
+        now.year(),
+        now.month(),
+        task_attempt.max(1),
+        file_attempt.max(1),
+        extension.trim_start_matches('.')
+    )
+}
+
 pub fn archive_key(retention_hours: i64, task_id: Uuid) -> String {
     let now = Utc::now();
     format!(
@@ -964,6 +984,8 @@ mod tests {
         assert!(key.starts_with("results/7d/"));
         assert!(key.ends_with("/00000000-0000-0000-0000-000000000001.webp"));
         assert!(archive_key(360, task_id).starts_with("archives/15d/"));
+        let attempt_key = result_attempt_key(24, task_id, file_id, 2, 3, "avif");
+        assert!(attempt_key.contains("-t2-f3.avif"));
     }
 
     #[test]